ai/gibidy
1
0
Fork 0
Code Activity

refactor: harden ChatGPT token lifecycle with startup recovery, single-writer locking, and faster auth flow

Browse source
This commit is contained in:
Arthur K. 2026-03-01 20:58:24 +03:00
parent 71d1050adb
commit 533e382e0e
Signed by: wzray
GPG key ID: B97F30FDC4636357
9 changed files with 313 additions and 178 deletions
Download patch file Download diff file Expand all files Collapse all files

250
src/providers/chatgpt/registration.py
View file

@ -78,6 +78,36 @@ def generate_name() -> str:
"Paul",
"Andrew",
"Joshua",
"Kenneth",
"Kevin",
"Brian",
"George",
"Edward",
"Ronald",
"Timothy",
"Jason",
"Jeffrey",
"Ryan",
"Jacob",
"Gary",
"Nicholas",
"Eric",
"Jonathan",
"Stephen",
"Larry",
"Justin",
"Scott",
"Brandon",
"Benjamin",
"Samuel",
"Frank",
"Gregory",
"Raymond",
"Alexander",
"Patrick",
"Jack",
"Dennis",
"Jerry",
]
last_names = [
"Smith",
@ -100,10 +130,47 @@ def generate_name() -> str:
"Moore",
"Jackson",
"Martin",
"Lee",
"Perez",
"Thompson",
"White",
"Harris",
"Sanchez",
"Clark",
"Ramirez",
"Lewis",
"Robinson",
"Walker",
"Young",
"Allen",
"King",
"Wright",
"Scott",
"Torres",
"Nguyen",
"Hill",
"Flores",
"Green",
"Adams",
"Nelson",
"Baker",
"Hall",
"Rivera",
"Campbell",
"Mitchell",
"Carter",
"Roberts",
]
return f"{random.choice(first_names)} {random.choice(last_names)}"
def generate_birthdate_90s() -> tuple[str, str, str]:
year = random.randint(1990, 1999)
month = random.randint(1, 12)
day = random.randint(1, 28)
return f"{month:02d}", f"{day:02d}", str(year)
def extract_verification_code(message: str) -> str | None:
normalized = re.sub(r"\s+", " ", message)
@ -177,24 +244,6 @@ async def exchange_code_for_tokens(code: str, verifier: str) -> ProviderTokens:
)
async def get_new_verification_code(
email_provider: BaseProvider,
email: str,
used_codes: set[str],
timeout_seconds: int = 240,
) -> str | None:
attempts = max(1, timeout_seconds // 5)
for _ in range(attempts):
message = await email_provider.get_latest_message(email)
if message:
all_codes = re.findall(r"\b(\d{6})\b", message)
for candidate in all_codes:
if candidate not in used_codes:
return candidate
await asyncio.sleep(5)
return None
async def get_latest_code(email_provider: BaseProvider, email: str) -> str | None:
message = await email_provider.get_latest_message(email)
if not message:
@ -209,24 +258,32 @@ async def fill_date_field(page: Page, month: str, day: str, year: str):
await month_field.scroll_into_view_if_needed()
await month_field.click()
await page.wait_for_timeout(120)
await page.wait_for_timeout(80)
await page.keyboard.type(f"{month}{day}{year}")
await page.wait_for_timeout(200)
await page.wait_for_timeout(120)
async def wait_for_signup_stabilization(page: Page):
try:
await page.wait_for_load_state("networkidle", timeout=15000)
except Exception:
logger.warning(
"Signup page did not reach networkidle quickly; continuing with fallback"
)
try:
await page.wait_for_load_state("domcontentloaded", timeout=10000)
except Exception:
pass
await page.wait_for_timeout(3000)
async def click_continue(page: Page, timeout_ms: int = 10000):
btn = page.get_by_role("button", name="Continue", exact=True).first
await btn.wait_for(state="visible", timeout=timeout_ms)
await btn.click()
async def wait_for_signup_stabilization(
page: Page,
source_url: str,
timeout_seconds: int = 30,
):
end_at = asyncio.get_running_loop().time() + timeout_seconds
while asyncio.get_running_loop().time() < end_at:
current_url = page.url
if current_url != source_url:
logger.info("Signup redirect detected: %s -> %s", source_url, current_url)
return
await asyncio.sleep(0.5)
logger.warning("Signup redirect was not detected within %ss", timeout_seconds)
async def register_chatgpt_account(
@ -238,7 +295,7 @@ async def register_chatgpt_account(
logger.error("No email provider factory configured")
return False
birth_month, birth_day, birth_year = "01", "15", "1995"
birth_month, birth_day, birth_year = generate_birthdate_90s()
current_page: Page | None = None
redirect_url_captured: str | None = None
@ -253,7 +310,7 @@ async def register_chatgpt_account(
)
email_provider = email_provider_factory(context)
logger.info("[1/6] Getting new email from configured provider...")
logger.info("[1/5] Getting new email from configured provider...")
email = await email_provider.get_new_email()
if not email:
raise AutomationError(
@ -266,127 +323,57 @@ async def register_chatgpt_account(
oauth_state = generate_state()
authorize_url = build_authorize_url(verifier, challenge, oauth_state)
logger.info("[2/6] Registering ChatGPT for %s", email)
logger.info("[2/5] Registering ChatGPT for %s", email)
chatgpt_page = await context.new_page()
current_page = chatgpt_page
await chatgpt_page.goto("https://chatgpt.com")
await chatgpt_page.wait_for_load_state("domcontentloaded")
await chatgpt_page.get_by_text("Sign up for free", exact=True).click()
await chatgpt_page.wait_for_timeout(2000)
await chatgpt_page.locator('input[type="email"]').first.wait_for(
state="visible", timeout=15000
)
await chatgpt_page.locator('input[type="email"]').fill(email)
await chatgpt_page.wait_for_timeout(500)
await chatgpt_page.get_by_role(
"button", name="Continue", exact=True
).click()
await chatgpt_page.wait_for_timeout(3000)
await click_continue(chatgpt_page)
await chatgpt_page.locator('input[type="password"]').first.wait_for(
state="visible", timeout=15000
)
await chatgpt_page.locator('input[type="password"]').fill(password)
await chatgpt_page.wait_for_timeout(500)
await chatgpt_page.get_by_role(
"button", name="Continue", exact=True
).click()
await chatgpt_page.wait_for_timeout(5000)
await click_continue(chatgpt_page)
await chatgpt_page.get_by_placeholder("Code").first.wait_for(
state="visible", timeout=30000
)
logger.info("[3/6] Getting verification message from email provider...")
logger.info("[3/5] Getting verification message from email provider...")
code = await get_latest_code(email_provider, email)
if not code:
raise AutomationError(
"email_provider", "Email provider returned no verification message"
)
logger.info("[3/6] Verification code extracted: %s", code)
used_codes = {code}
logger.info("[3/5] Verification code extracted: %s", code)
await chatgpt_page.bring_to_front()
code_input = chatgpt_page.get_by_placeholder("Code")
if await code_input.count() > 0:
await code_input.fill(code)
await chatgpt_page.wait_for_timeout(5000)
await click_continue(chatgpt_page)
continue_btn = chatgpt_page.get_by_role(
"button", name="Continue", exact=True
)
if await continue_btn.count() > 0:
await continue_btn.click()
await chatgpt_page.wait_for_timeout(5000)
logger.info("[4/6] Setting profile...")
logger.info("[4/5] Setting profile...")
name_input = chatgpt_page.get_by_placeholder("Full name")
await name_input.first.wait_for(state="visible", timeout=20000)
if await name_input.count() > 0:
await name_input.fill(full_name)
await chatgpt_page.wait_for_timeout(500)
await fill_date_field(chatgpt_page, birth_month, birth_day, birth_year)
await chatgpt_page.wait_for_timeout(1000)
continue_btn = chatgpt_page.get_by_role(
"button", name="Continue", exact=True
)
if await continue_btn.count() > 0:
await continue_btn.click()
profile_url = chatgpt_page.url
await click_continue(chatgpt_page)
logger.info("Account registered!")
await wait_for_signup_stabilization(chatgpt_page)
await wait_for_signup_stabilization(chatgpt_page, source_url=profile_url)
logger.info("[5/6] Skipping onboarding...")
for _ in range(5):
skip_btn = chatgpt_page.locator(
'button:has-text("Skip"):not(:has-text("Skip Tour"))'
)
if await skip_btn.count() > 0:
for i in range(await skip_btn.count()):
try:
btn = skip_btn.nth(i)
if await btn.is_visible():
await btn.click(timeout=5000)
logger.info("Clicked: Skip")
await chatgpt_page.wait_for_timeout(1500)
except:
pass
await chatgpt_page.wait_for_timeout(1000)
skip_tour = chatgpt_page.locator('button:has-text("Skip Tour")')
if await skip_tour.count() > 0:
try:
await skip_tour.first.wait_for(state="visible", timeout=5000)
await skip_tour.first.click(timeout=5000)
logger.info("Clicked: Skip Tour")
await chatgpt_page.wait_for_timeout(2000)
except:
pass
await chatgpt_page.wait_for_timeout(2000)
for _ in range(3):
continue_btn = chatgpt_page.locator('button:has-text("Continue")')
if await continue_btn.count() > 0:
try:
await continue_btn.first.wait_for(state="visible", timeout=5000)
await continue_btn.first.click(timeout=5000)
logger.info("Clicked: Continue")
await chatgpt_page.wait_for_timeout(2000)
except:
pass
await chatgpt_page.wait_for_timeout(2000)
okay_btn = chatgpt_page.locator('button:has-text("Okay, let")')
for _ in range(10):
try:
await okay_btn.first.wait_for(state="visible", timeout=3000)
await okay_btn.first.click(timeout=5000)
logger.info("Clicked: Okay, let's go")
await chatgpt_page.wait_for_timeout(3000)
break
except:
await chatgpt_page.wait_for_timeout(1000)
logger.info("Skipping subscription/card flow (disabled)")
await chatgpt_page.wait_for_timeout(2000)
logger.info("[6/6] Running OAuth flow to get tokens...")
logger.info("[5/5] Running OAuth flow to get tokens...")
oauth_page = await context.new_page()
current_page = oauth_page
@ -400,33 +387,34 @@ async def register_chatgpt_account(
oauth_page.on("request", handle_request)
await oauth_page.goto(authorize_url, wait_until="domcontentloaded")
await oauth_page.wait_for_timeout(2000)
await oauth_page.locator(
'input[type="email"], input[name="email"]'
).first.wait_for(state="visible", timeout=20000)
email_input = oauth_page.locator('input[type="email"], input[name="email"]')
if await email_input.count() > 0:
await email_input.first.fill(email)
await oauth_page.wait_for_timeout(400)
continue_button = oauth_page.get_by_role("button", name="Continue")
if await continue_button.count() > 0:
await continue_button.first.click()
await oauth_page.wait_for_timeout(2500)
await oauth_page.locator('input[type="password"]').first.wait_for(
state="visible", timeout=20000
)
password_input = oauth_page.locator('input[type="password"]')
if await password_input.count() > 0:
await password_input.first.fill(password)
await oauth_page.wait_for_timeout(400)
continue_button = oauth_page.get_by_role("button", name="Continue")
if await continue_button.count() > 0:
await continue_button.first.click()
await oauth_page.wait_for_timeout(2500)
for label in ["Continue", "Allow", "Authorize"]:
button = oauth_page.get_by_role("button", name=label)
if await button.count() > 0:
try:
await button.first.click(timeout=5000)
await oauth_page.wait_for_timeout(2000)
await oauth_page.wait_for_timeout(500)
except Exception:
pass