"""Trikatuka playlist-transfer driver. The site is an Angular SPA at https://trikatuka.aknakn.eu/#/. Two buttons trigger OAuth popup windows for user1 (old account) and user2 (new account). After both log in, a "Transfer all" button copies user1's playlists + premium to user2. We fill each OAuth popup via window-switching in SeleniumBase. """ import time from typing import Any, Dict, Optional import abort from config import Config from utils.fill import ( click_by_text, fill_instant, type_into, wait_for_selector, ) from utils.logger import child log = child("trikatuka") TRIKATUKA_URL = "https://trikatuka.aknakn.eu/#/" def _wait_login_button(sb, which: str, timeout_s: float = 30) -> Optional[str]: """`which` is 'user1' or 'user2'. Returns the matched selector or None.""" sel_pairs = { "user1": [ 'a[ng-click="user1.login()"]', 'button[ng-click="user1.login()"]', ], "user2": [ 'a[ng-click="user2.login()"]', 'button[ng-click="user2.login()"]', ], } return wait_for_selector(sb.driver, sel_pairs[which], timeout_s=timeout_s) def _click_selector(sb, selector: str) -> bool: try: return bool( sb.driver.execute_script( "const n=document.querySelector(arguments[0]); if(n){ n.click(); return true; } return false;", selector, ) ) except Exception: return False def _fill_oauth_popup(sb, email: str, password: str, timeout_s: float = 30) -> Dict[str, Any]: """Detect the newly-opened Spotify OAuth popup window, fill it, submit.""" driver = sb.driver main_handle = driver.current_window_handle original_handles = set(driver.window_handles) deadline = time.monotonic() + timeout_s popup_handle = None while time.monotonic() < deadline: abort.check() new_handles = set(driver.window_handles) - original_handles if new_handles: popup_handle = next(iter(new_handles)) break time.sleep(0.25) if not popup_handle: return {"ok": False, "error": "OAuth popup did not open"} try: driver.switch_to.window(popup_handle) # Email / username — some variants prefill if cookies remember email_sel = wait_for_selector( driver, [ "input#username", 'input[data-testid="login-username"]', 'input[autocomplete="username"]', 'input[name="username"]', 'input[type="email"]', ], timeout_s=12, ) if email_sel: # If prefilled with the same email, skip; else set current = driver.execute_script("const e=document.querySelector(arguments[0]); return e ? e.value : '';", email_sel) if (current or "").strip().lower() != email.strip().lower(): fill_instant(driver, email_sel, email) pw_sel = wait_for_selector( driver, [ "input#password", 'input[data-testid="login-password"]', 'input[type="password"]', ], timeout_s=12, ) if not pw_sel: return {"ok": False, "error": "password field not found in popup"} type_into(driver, pw_sel, password, per_char_delay_ms=25) time.sleep(0.6) # Click Log in clicked = False for sel in ( 'button[type="submit"]', 'button[data-testid="login-button"]', 'button[data-encore-id="buttonPrimary"]', ): if _click_selector(sb, sel): clicked = True break if not clicked: click_by_text(driver, "Log in") or click_by_text(driver, "Sign in") # Wait for popup to close OR URL to navigate away from /login start_url = driver.current_url close_deadline = time.monotonic() + 45 while time.monotonic() < close_deadline: abort.check() handles_now = driver.window_handles if popup_handle not in handles_now: break try: if driver.current_window_handle == popup_handle: if "/login" not in (driver.current_url or "") and driver.current_url != start_url: break except Exception: break time.sleep(0.5) finally: # Return to main window regardless try: if popup_handle in driver.window_handles: driver.switch_to.window(popup_handle) driver.close() except Exception: pass driver.switch_to.window(main_handle) return {"ok": True} def _wait_transfer_button(sb, timeout_s: float = 60) -> Optional[str]: """Find any visible clickable whose text contains 'transfer'.""" deadline = time.monotonic() + timeout_s js = r""" (function() { const nodes = document.querySelectorAll('button, a[ng-click], [role="button"]'); for (const el of nodes) { const r = el.getBoundingClientRect(); const s = window.getComputedStyle(el); if (s.display === 'none' || s.visibility === 'hidden' || parseFloat(s.opacity) === 0) continue; if (r.width === 0 && r.height === 0) continue; const t = (el.innerText || '').trim().toLowerCase(); if (t.includes('transfer')) { el.setAttribute('data-ffsb-transfer', '1'); return '[data-ffsb-transfer="1"]'; } } return null; })(); """ while time.monotonic() < deadline: abort.check() try: sel = sb.driver.execute_script(js) if sel: return sel except Exception: pass time.sleep(0.5) return None def run(sb, params: Dict[str, Any], cfg: Config) -> Dict[str, Any]: old_email = params["old_email"] old_password = params["old_password"] new_email = params["new_email"] new_password = params["new_password"] log.info(f"trikatuka: opening {TRIKATUKA_URL}") sb.open(TRIKATUKA_URL) time.sleep(2.0) u1 = _wait_login_button(sb, "user1", timeout_s=30) if not u1: return {"ok": False, "error": "trikatuka user1 login button not found"} _click_selector(sb, u1) log.info("clicked user1 login; filling OAuth popup") r = _fill_oauth_popup(sb, old_email, old_password) if not r.get("ok"): return {"ok": False, "error": "user1 oauth: " + r.get("error", "unknown")} time.sleep(2.0) u2 = _wait_login_button(sb, "user2", timeout_s=20) if not u2: return {"ok": False, "error": "trikatuka user2 login button not found"} _click_selector(sb, u2) log.info("clicked user2 login; filling OAuth popup") r = _fill_oauth_popup(sb, new_email, new_password) if not r.get("ok"): return {"ok": False, "error": "user2 oauth: " + r.get("error", "unknown")} log.info("awaiting Transfer all button") xfer = _wait_transfer_button(sb, timeout_s=60) if not xfer: return {"ok": False, "error": "Transfer all button not found"} _click_selector(sb, xfer) log.info("clicked Transfer all") time.sleep(10.0) return {"ok": True} def run_standalone(sb_factory, params: Dict[str, Any], cfg: Config) -> Dict[str, Any]: with sb_factory() as sb: abort.register_sb(sb) try: return run(sb, params, cfg) finally: abort.clear_sb()