God-Tier Python: Extracting YouTube Channel Business Emails Without Touching the CAPTCHA

import asyncio import json import re from urllib.parse import unquote, urlparse import httpx EMAIL_REGEX = re.compile( r"[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+", re.IGNORECASE ) # Exclude common static-asset fake emails INVALID_EXTENSIONS = ( ".png", ".jpg", ".jpeg", ".gif", ".webp", ".svg", ".css", ".js", ) EXCLUDED_EMAILS = {"support@youtube.com", "privacy@google.com"} HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36", "Accept-Language": "en-US,en;q=0.9", } def clean_emails(raw_emails: list[str]) -> list[str]: valid = set() for e in raw_emails: e = e.strip().lower() if any(e.endswith(ext) for ext in INVALID_EXTENSIONS): continue if e in EXCLUDED_EMAILS: continue valid.add(e) return list(valid) class YouTubeChannelLeadExtractor: def __init__(self, concurrency: int = 5): self.semaphore = asyncio.Semaphore(concurrency) self.client = httpx.AsyncClient( headers=HEADERS, timeout=12.0, follow_redirects=True ) async def _fetch_external_site_email(self, url: str) -> list[str]: """Shallow-scan a Linktree, personal site or Shopify page for a contact email""" try: parsed = urlparse(url) if any( domain in parsed.netloc for domain in ["instagram.com", "tiktok.com", "facebook.com"] ): return [] resp = await self.client.get(url) if resp.status_code == 200: return EMAIL_REGEX.findall(resp.text) except Exception: pass return [] async def extract_from_channel(self, channel_url: str) -> dict: """Reverse-parse a single channel page automatically""" async with self.semaphore: target_url = channel_url.rstrip("/") + "/about" result = { "channel_url": channel_url, "channel_name": "Unknown", "emails": [], "source": "None", "external_links": [], } try: resp = await self.client.get(target_url) if resp.status_code != 200: return result html = resp.text # 1. Grab the ytInitialData context variable (YouTube's first-screen JSON)  json_match = re.search( r"var ytInitialData\s*=\s*({.+?});</script>", html ) if not json_match: json_match = re.search( r"window\[\"ytInitialData\"\]\s*=\s*({.+?});", html ) if json_match: data = json.loads(json_match.group(1)) try: result["channel_name"] = data["metadata"][ "channelMetadataRenderer" ]["title"] except KeyError: pass raw_data_str = json.dumps(data) found_redirect_links = re.findall( r"https://www.youtube.com/redirect\?event=channel_description[^\"\s]+", raw_data_str, ) for r_link in found_redirect_links: match_q = re.search(r"[?&]q=([^&]+)", r_link) if match_q: decoded = unquote(match_q.group(1)) if ( decoded.startswith("http") and decoded not in result["external_links"] ): result["external_links"].append(decoded) # 2. Strategy A: regex the HTML and description metadata directly  direct_emails = EMAIL_REGEX.findall(html) clean_direct = clean_emails(direct_emails) if clean_direct: result["emails"] = clean_direct result["source"] = "Channel Description / Metadata" return result # 3. Strategy B: if no plaintext in the bio, deep-scan external links concurrently  if result["external_links"]: scan_tasks = [ self._fetch_external_site_email(link) for link in result["external_links"][:3] # first 3 links only  ] external_results = await asyncio.gather(*scan_tasks) combined_external = [] for emails in external_results: combined_external.extend(emails) clean_external = clean_emails(combined_external) if clean_external: result["emails"] = clean_external result["source"] = "External Links (Website/Linktree)" return result except Exception as err: result["error"] = str(err) return result async def close(self): await self.client.aclose() # ==================== Test run ==================== async def main(): # Supports handle URLs and channel-ID URLs  test_channels = [ "https://www.youtube.com/@mkbhd", "https://www.youtube.com/@LinusTechTips", "https://www.youtube.com/@ExplosionExplosive", ] extractor = YouTubeChannelLeadExtractor(concurrency=3) tasks = [extractor.extract_from_channel(ch) for ch in test_channels] leads = await asyncio.gather(*tasks) await extractor.close() print(json.dumps(leads, indent=2, ensure_ascii=False)) if __name__ == "__main__": asyncio.run(main())