# backend/tools/social_tools.py # Returns structured status keys so the React layer can use i18n.t() for display text. import logging import asyncio from backend.tools.browser_tools import get_ephemeral_context, stealth_async from backend.tools.captcha_solver import solve_captcha_if_present def _get_vault_credential(site_like: str): """Retrieve AES decrypted credentials from Vault DB.""" from backend.services.usb_vault import get_secret if "x.com" in site_like or "twitter" in site_like: return get_secret("X_EMAIL"), get_secret("X_PASS") elif "google.com" in site_like or "youtube" in site_like: return get_secret("YOUTUBE_EMAIL"), get_secret("YOUTUBE_PASS") elif "insta" in site_like: return get_secret("INSTA_USER"), get_secret("INSTA_PASS") return None, None async def login_to_socials() -> dict: """ Initial setup routine. Navigates to X and YouTube. If auth needed, auto-injects credentials from Vault. Returns status_key for i18n rendering. """ try: async with get_ephemeral_context() as context: page = await context.new_page() if stealth_async: await stealth_async(page) # 1. Check X (Twitter) await page.goto("https://x.com/home") await page.wait_for_timeout(3000) await solve_captcha_if_present(page) x_title = await page.title() if "log in" in x_title.lower() or "login" in x_title.lower(): user, pwd = _get_vault_credential("x.com") if user and pwd: try: await page.goto("https://x.com/i/flow/login") await page.wait_for_selector('input[autocomplete="username"]', timeout=5000) await page.fill('input[autocomplete="username"]', user) await page.keyboard.press('Enter') await page.wait_for_selector('input[type="password"]', timeout=5000) await page.fill('input[type="password"]', pwd) await page.keyboard.press('Enter') await page.wait_for_timeout(5000) await solve_captcha_if_present(page) except Exception as e: logging.error(f"X auto-login failed: {e}") # 2. Check YouTube await page.goto("https://youtube.com") await page.wait_for_timeout(3000) await solve_captcha_if_present(page) yt_title = await page.title() # YouTube auto-login if "sign in" in yt_title.lower(): user, pwd = _get_vault_credential("google.com") if user and pwd: try: await page.goto("https://accounts.google.com/ServiceLogin") await page.wait_for_selector('input[type="email"]', timeout=5000) await page.fill('input[type="email"]', user) await page.keyboard.press('Enter') await page.wait_for_timeout(3000) await page.wait_for_selector('input[type="password"]', timeout=5000) await page.fill('input[type="password"]', pwd) await page.keyboard.press('Enter') await page.wait_for_timeout(5000) await solve_captcha_if_present(page) except Exception as e: logging.error(f"YouTube auto-login failed: {e}") # 3. Check Instagram await page.goto("https://instagram.com") await page.wait_for_timeout(3000) await solve_captcha_if_present(page) insta_title = await page.title() # Instagram auto-login if "login" in insta_title.lower() or "instagram" in insta_title.lower(): user, pwd = _get_vault_credential("insta") if user and pwd: try: # Try to see if login form is present form_present = await page.evaluate('() => document.querySelector("input[name="username"]") !== null') if form_present: await page.fill('input[name="username"]', user) await page.fill('input[name="password"]', pwd) await page.keyboard.press('Enter') await page.wait_for_timeout(5000) await solve_captcha_if_present(page) except Exception as e: logging.error(f"Instagram auto-login failed: {e}") # Final check await page.goto("https://x.com/home") await page.wait_for_timeout(2000) x_title_final = await page.title() await page.goto("https://youtube.com") await page.wait_for_timeout(2000) yt_title_final = await page.title() await page.goto("https://instagram.com") await page.wait_for_timeout(2000) insta_title_final = await page.title() await page.close() needs_auth = "log in" in x_title_final.lower() or "sign in" in yt_title_final.lower() or "login" in insta_title_final.lower() return { "status": "ok", "status_key": "social.login_needs_auth" if needs_auth else "social.login_success", "x_title": x_title_final, "yt_title": yt_title_final, "insta_title": insta_title_final, } except Exception as e: logging.error(f"Social login error: {e}") return {"status": "error", "status_key": "social.post_failed", "detail": str(e)} async def post_to_x(tweet_text: str) -> dict: try: async with get_ephemeral_context() as context: page = await context.new_page() if stealth_async: await stealth_async(page) await page.goto("https://x.com/compose/tweet") await page.wait_for_selector('div[data-testid="tweetTextarea_0"]', timeout=10000) await solve_captcha_if_present(page) await page.fill('div[data-testid="tweetTextarea_0"]', tweet_text) await page.click('div[data-testid="tweetButton"]') await page.wait_for_timeout(3000) await page.close() return {"status": "ok", "status_key": "social.post_success", "preview": tweet_text[:30]} except Exception as e: return {"status": "error", "status_key": "social.post_failed", "detail": str(e)} async def scrape_instagram(url: str) -> dict: """ Scrape Instagram using 429 exponential backoff + Playwright fallback. """ import httpx max_retries = 3 base_delay = 2 # Attempt 1: Fast HTTPx request (often gets 429) for attempt in range(max_retries): try: async with httpx.AsyncClient() as client: res = await client.get(url, headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"}) if res.status_code == 429: delay = base_delay * (2 ** attempt) logging.warning(f"Instagram 429 Rate Limit. Backing off for {delay}s...") await asyncio.sleep(delay) continue if res.status_code == 200: return {"status": "success", "source": "httpx", "content": res.text[:2000]} break except Exception as e: logging.error(f"httpx instagram failed: {e}") break # Attempt 2: Playwright fallback logging.info("Falling back to Playwright for Instagram...") try: async with get_ephemeral_context() as context: page = await context.new_page() if stealth_async: await stealth_async(page) await page.goto(url) await page.wait_for_timeout(4000) await solve_captcha_if_present(page) content = await page.content() await page.close() return {"status": "success", "source": "playwright", "content": content[:2000]} except Exception as e: return {"status": "error", "detail": str(e)}