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 = {"[email protected]", "[email protected]"}

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())