feat: add 5 new email_scan modules entertainment(1), learning(1), other(2), social(1)

user_scanner/core/helpers.py

@@ -37,6 +37,9 @@
         "hellochinese",
         "hanzii",
         "programminghub",
+        "talkpal",
+        "dragongroot",
+        "hoichoi",
     ],
 }
 

user_scanner/email_scan/entertainment/hoichoi.py

@@ -0,0 +1,68 @@
+import httpx
+import random
+import string
+from user_scanner.core.result import Result
+
+
+def _generate_device_id() -> str:
+    """Generate a random 16-character hex string for device ID."""
+    return "".join(random.choices(string.hexdigits.lower(), k=16))
+
+
+async def _check(email: str) -> Result:
+    url = "https://prod-api.hoichoi.dev/core/api/v1/auth/signinup/code"
+    show_url = "https://www.hoichoi.tv"
+
+    device_id = _generate_device_id()
+
+    params = {
+        'platform': "ANDROID_MOBILE",
+        'language': "english",
+        'appVersion': "3.1.44",
+        'deviceId': device_id
+    }
+
+    payload = {
+        "email": email,
+        "deviceId": device_id
+    }
+
+    headers = {
+        'User-Agent': "Mozilla/5.0 (Linux; Android 13; wv) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/149.0.0.0 Mobile Safari/537.36",
+        'Accept': "application/json",
+        'Accept-Encoding': "gzip;q=1.0,deflate;q=0.9",
+        'Content-Type': "application/json",
+        'x-hoichoi-siteid': "hoichoitv",
+        'x-bypass-proxy': "true"
+    }
+
+    async with httpx.AsyncClient(http2=True) as client:
+        try:
+            response = await client.post(url, params=params, json=payload, headers=headers, timeout=6.0)
+
+            if response.status_code == 429:
+                return Result.error("Rate limited", url=show_url)
+
+            if response.status_code == 400:
+                data = response.json()
+                if data.get("errorCode") == "EMAIL_SIGNUP_NOT_SUPPORTED":
+                    return Result.available(url=show_url)
+                return Result.error("Unexpected response body, report it via GitHub issues", url=show_url)
+
+            if response.status_code == 200:
+                data = response.json()
+                if data.get("status") == "OK" and data.get("flowType") == "USER_INPUT_CODE":
+                    return Result.taken(url=show_url)
+                return Result.error("Unexpected response body, report it via GitHub issues", url=show_url)
+
+            return Result.error(f"Unexpected response status: {response.status_code}, report it via GitHub issues", url=show_url)
+
+        except Exception as e:
+            return Result.error(e, url=show_url)
+
+
+async def validate_hoichoi(email: str) -> Result:
+    """
+    Hoichoi email validator. Note: This will send a sign-in OTP email if it exists.
+    """
+    return await _check(email)

user_scanner/email_scan/learning/talkpal.py

@@ -0,0 +1,52 @@
+import httpx
+from user_scanner.core.result import Result
+
+
+async def _check(email: str) -> Result:
+    url = "https://api.talkpal.ai/api/v1/auth/recovery-request"
+    show_url = "https://talkpal.ai"
+
+    payload = {
+        "email": email,
+        "source": "app"
+    }
+
+    headers = {
+        'User-Agent': "NitroFetch/1.0",
+        'Accept': "application/json, text/plain, */*",
+        'Content-Type': "application/json",
+        'source': "app",
+        'source-device': "ANDROID",
+        'priority': "u=1, i"
+    }
+
+    async with httpx.AsyncClient(http2=True) as client:
+        try:
+            response = await client.post(url, json=payload, headers=headers, timeout=6.0)
+
+            if response.status_code == 429:
+                return Result.error("Rate limited", url=show_url)
+
+            if response.status_code == 404:
+                data = response.json()
+                if data.get("message") == "User not found" and data.get("statusCode") == 404:
+                    return Result.available(url=show_url)
+                return Result.error("Unexpected response body, report it via GitHub issues", url=show_url)
+
+            if response.status_code == 201:
+                data = response.json()
+                if data.get("status") is True and "Password reset email has been sent" in data.get("message", ""):
+                    return Result.taken(url=show_url)
+                return Result.error("Unexpected response body, report it via GitHub issues", url=show_url)
+
+            return Result.error(f"Unexpected response status: {response.status_code}, report it via GitHub issues", url=show_url)
+
+        except Exception as e:
+            return Result.error(e, url=show_url)
+
+
+async def validate_talkpal(email: str) -> Result:
+    """
+    TalkPal email validator. Note: This will send a password reset email if it exists.
+    """
+    return await _check(email)

user_scanner/email_scan/other/dragongroot.py

@@ -0,0 +1,49 @@
+import httpx
+from user_scanner.core.result import Result
+
+
+async def _check(email: str) -> Result:
+    url = "https://hayat.dragongroot.com/api/auth/signup/send-otp"
+    show_url = "https://dragongroot.com"
+
+    payload = {
+        "email": email
+    }
+
+    headers = {
+        'User-Agent': "okhttp/4.12.0",
+        'Accept': "application/json",
+        'Accept-Encoding': "gzip",
+        'Content-Type': "application/json"
+    }
+
+    async with httpx.AsyncClient(http2=True) as client:
+        try:
+            response = await client.post(url, json=payload, headers=headers, timeout=6.0)
+
+            if response.status_code == 429:
+                return Result.error("Rate limited", url=show_url)
+
+            if response.status_code == 400:
+                data = response.json()
+                if data.get("success") is False and data.get("message") == "An account with this email already exists":
+                    return Result.taken(url=show_url)
+                return Result.error("Unexpected response body, report it via GitHub issues", url=show_url)
+
+            if response.status_code == 200:
+                data = response.json()
+                if data.get("success") is True and data.get("message") == "OTP sent to your email":
+                    return Result.available(url=show_url)
+                return Result.error("Unexpected response body, report it via GitHub issues", url=show_url)
+
+            return Result.error(f"Unexpected response status: {response.status_code}, report it via GitHub issues", url=show_url)
+
+        except Exception as e:
+            return Result.error(e, url=show_url)
+
+
+async def validate_dragongroot(email: str) -> Result:
+    """
+    Dragongroot email validator. Note: This will send a signup OTP email if it is available.
+    """
+    return await _check(email)

user_scanner/email_scan/other/numsify.py

@@ -0,0 +1,41 @@
+import httpx
+from user_scanner.core.result import Result
+async def _check(email: str) -> Result:
+    url = "https://api.numiva.me/api/auth/register"
+    show_url = "https://numsify.com"
+    payload = {
+        "email": email,
+        "password": "everyone_thought_i_am_heartless",  # Intentionally fails strength check (no number)
+        "first_name": "theknight",
+        "last_name": "nevergivesup"
+    }
+    headers = {
+        'User-Agent': "Dart/3.11 (dart:io)",
+        'Accept': "application/json",
+        'Accept-Encoding': "gzip",
+        'Content-Type': "application/json",
+        'x-device-name': "Android:google-Pixel 6-13",
+        'lang': "en",
+        'authorization': "Bearer null",
+        'x-app-version': "1.4.53+73"
+    }
+    async with httpx.AsyncClient(http2=True) as client:
+        try:
+            response = await client.post(url, json=payload, headers=headers, timeout=6.0)
+            if response.status_code == 429:
+                return Result.error("Rate limited", url=show_url)
+            data = response.json()
+            error_data = data.get("error", {})
+            error_msg = error_data.get("message", "")
+            if "password must contain at least one number" in error_msg:
+                return Result.available(url=show_url)
+            elif "email already registered" in error_msg:
+                return Result.taken(url=show_url)
+            return Result.error("Unexpected response body, report it via GitHub issues", url=show_url)
+        except Exception as e:
+            return Result.error(e, url=show_url)
+async def validate_numsify(email: str) -> Result:
+    """
+    Numsify email validator. Checks registration endpoint without sending an email.
+    """
+    return await _check(email)

user_scanner/email_scan/social/gravatar.py

@@ -0,0 +1,66 @@
+import hashlib
+import httpx
+from user_scanner.core.result import Result
+async def _check(email: str) -> Result:
+    show_url = "https://gravatar.com"
+    email_clean = email.lower().strip()
+    email_hash = hashlib.sha256(email_clean.encode("utf-8")).hexdigest()
+    url = f"https://www.gravatar.com/avatar/{email_hash}?d=404"
+    headers = {
+        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
+    }
+    try:
+        async with httpx.AsyncClient(timeout=5.0) as client:
+            response = await client.get(url, headers=headers)
+            if response.status_code == 200:
+                extra = {"avatar_url": f"https://www.gravatar.com/avatar/{email_hash}"}
+                # Optionally attempt to load public profile data for enriched metadata
+                profile_url = f"https://en.gravatar.com/{email_hash}.json"
+                try:
+                    profile_resp = await client.get(profile_url, headers=headers, timeout=3.0)
+                    if profile_resp.status_code == 200:
+                        data = profile_resp.json()
+                        entry = data.get("entry", [{}])[0]
+                        if entry.get("preferredUsername"):
+                            extra["username"] = entry["preferredUsername"]
+                        if entry.get("displayName"):
+                            extra["display_name"] = entry["displayName"]
+                        if entry.get("profileUrl"):
+                            extra["profile_url"] = entry["profileUrl"]
+                except Exception:
+                    pass
+                return Result.taken(url=show_url, extra=extra)
+            elif response.status_code == 404:
+                # Also fall back to check MD5 since some older profiles might only map via MD5
+                email_md5 = hashlib.md5(email_clean.encode("utf-8")).hexdigest()
+                url_md5 = f"https://www.gravatar.com/avatar/{email_md5}?d=404"
+
+                response_md5 = await client.get(url_md5, headers=headers)
+                if response_md5.status_code == 200:
+                    extra = {"avatar_url": f"https://www.gravatar.com/avatar/{email_md5}"}
+                    profile_url_md5 = f"https://en.gravatar.com/{email_md5}.json"
+                    try:
+                        profile_resp = await client.get(profile_url_md5, headers=headers, timeout=3.0)
+                        if profile_resp.status_code == 200:
+                            data = profile_resp.json()
+                            entry = data.get("entry", [{}])[0]
+                            if entry.get("preferredUsername"):
+                                extra["username"] = entry["preferredUsername"]
+                            if entry.get("displayName"):
+                                extra["display_name"] = entry["displayName"]
+                            if entry.get("profileUrl"):
+                                extra["profile_url"] = entry["profileUrl"]
+                    except Exception:
+                        pass
+                    return Result.taken(url=show_url, extra=extra)
+                elif response_md5.status_code == 404:
+                    return Result.available(url=show_url)
+                else:
+                    return Result.error(f"HTTP MD5 {response_md5.status_code}")
+            return Result.error(f"HTTP {response.status_code}")
+    except httpx.TimeoutException:
+        return Result.error("Connection timed out")
+    except Exception as e:
+        return Result.error(e)
+async def validate_gravatar(email: str) -> Result:
+    return await _check(email)