242 lines
No EOL
9 KiB
Python
242 lines
No EOL
9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Simple WhatsApp Discovery Test - 10 variations only
|
|
|
|
Tests 10 variations of a phone number using WhatsApp Business API.
|
|
Stores raw API responses for further analysis.
|
|
|
|
Usage:
|
|
python whatsapp_discovery_simple.py <phone_base> <country_code>
|
|
|
|
Example:
|
|
python whatsapp_discovery_simple.py 3162940 31
|
|
"""
|
|
|
|
import asyncio
|
|
import aiohttp
|
|
import json
|
|
import os
|
|
import sys
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
class SimpleWhatsAppDiscovery:
|
|
"""Tests 10 variations of a phone number."""
|
|
|
|
def __init__(self, phone_base: str, country_code: str):
|
|
self.phone_base = phone_base.replace('XXXX', '').replace('X', '')
|
|
self.country_code = country_code
|
|
self.results = []
|
|
|
|
async def test_single_number(self, session, phone: str, attempt_num: int):
|
|
"""Test a single phone number."""
|
|
print(f" Testing {phone} (attempt {attempt_num})...")
|
|
|
|
# Check for API token
|
|
if not os.getenv("WHATSAPP_BUSINESS_TOKEN"):
|
|
return {
|
|
"phone": phone,
|
|
"attempt": attempt_num,
|
|
"status": "error",
|
|
"error": "WHATSAPP_BUSINESS_TOKEN environment variable not set",
|
|
"raw_response": None
|
|
}
|
|
|
|
headers = {
|
|
'Authorization': f'Bearer {os.getenv("WHATSAPP_BUSINESS_TOKEN")}',
|
|
'Content-Type': 'application/json'
|
|
}
|
|
|
|
# WhatsApp Business API endpoint
|
|
api_url = f"https://graph.facebook.com/v18.0/phone_numbers"
|
|
|
|
data = {
|
|
"phone_number": phone,
|
|
"fields": ["display_name", "is_business", "last_activity"]
|
|
}
|
|
|
|
try:
|
|
async with session.post(api_url, json=data, headers=headers) as response:
|
|
response_text = await response.text()
|
|
|
|
result = {
|
|
"phone": phone,
|
|
"attempt": attempt_num,
|
|
"status_code": response.status,
|
|
"status": "unknown",
|
|
"raw_response": response_text,
|
|
"headers": dict(response.headers)
|
|
}
|
|
|
|
# Parse status
|
|
if response.status == 200:
|
|
result["status"] = "success"
|
|
try:
|
|
result_data = json.loads(response_text)
|
|
if result_data.get("data"):
|
|
phone_info = result_data["data"][0]
|
|
result["exists"] = True
|
|
result["display_name"] = phone_info.get("display_name")
|
|
result["is_business"] = phone_info.get("is_business", False)
|
|
result["last_activity"] = phone_info.get("last_activity")
|
|
else:
|
|
result["exists"] = False
|
|
result["message"] = "No data returned"
|
|
except json.JSONDecodeError:
|
|
result["parse_error"] = "Failed to parse JSON response"
|
|
|
|
elif response.status == 404:
|
|
result["status"] = "not_found"
|
|
result["exists"] = False
|
|
|
|
else:
|
|
result["status"] = "error"
|
|
result["exists"] = False
|
|
result["error"] = f"HTTP {response.status}"
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
return {
|
|
"phone": phone,
|
|
"attempt": attempt_num,
|
|
"status": "exception",
|
|
"error": str(e),
|
|
"exists": False,
|
|
"raw_response": None
|
|
}
|
|
|
|
def generate_10_variations(self):
|
|
"""Generate 10 variations: original + 9 with last digit changed."""
|
|
base = self.phone_base
|
|
variations = []
|
|
|
|
# Original
|
|
variations.append(base)
|
|
|
|
# Replace last digit with 0-9
|
|
if len(base) >= 1:
|
|
for i in range(10):
|
|
variation = base[:-1] + str(i)
|
|
variations.append(variation)
|
|
|
|
# Return unique variations only (remove duplicates)
|
|
return list(dict.fromkeys(variations))[:10]
|
|
|
|
async def run_test(self):
|
|
"""Run test on 10 variations."""
|
|
full_phone = f"{self.country_code}{self.phone_base}XXXX"
|
|
print(f"🔍 Testing WhatsApp for {full_phone}")
|
|
print(f"📊 Testing 10 variations using last digit replacement")
|
|
|
|
variations = self.generate_10_variations()
|
|
print(f"📈 Generated {len(variations)} variations")
|
|
|
|
# Create HTTP session
|
|
timeout = aiohttp.ClientTimeout(total=30)
|
|
connector = aiohttp.TCPConnector(limit=10, force_close=True)
|
|
|
|
async with aiohttp.ClientSession(
|
|
timeout=timeout,
|
|
connector=connector,
|
|
headers={'User-Agent': 'WhatsApp-Discovery-Test/1.0'}
|
|
) as session:
|
|
|
|
# Test each variation
|
|
for i, variation in enumerate(variations):
|
|
phone = f"{self.country_code}{variation}"
|
|
result = await self.test_single_number(session, phone, i + 1)
|
|
self.results.append(result)
|
|
|
|
# Print immediate result
|
|
if result.get("exists"):
|
|
print(f" ✅ {phone} -> FOUND! Display: {result.get('display_name', 'N/A')}")
|
|
elif result["status"] == "not_found":
|
|
print(f" ❌ {phone} -> Not found")
|
|
elif result["status"] == "success" and not result.get("exists"):
|
|
print(f" ⚪ {phone} -> No WhatsApp account")
|
|
else:
|
|
print(f" ⚠️ {phone} -> {result['status']}: {result.get('error', 'Unknown error')}")
|
|
|
|
# Small delay between requests
|
|
await asyncio.sleep(0.5)
|
|
|
|
# Summary
|
|
found_count = sum(1 for r in self.results if r.get("exists"))
|
|
print(f"\n📊 Test Complete!")
|
|
print(f" Total tested: {len(self.results)}")
|
|
print(f" WhatsApp accounts found: {found_count}")
|
|
|
|
if found_count > 0:
|
|
print(f"\n🎯 Found WhatsApp Accounts:")
|
|
for result in self.results:
|
|
if result.get("exists"):
|
|
print(f" 📱 {result['phone']}")
|
|
print(f" Name: {result.get('display_name', 'N/A')}")
|
|
print(f" Business: {result.get('is_business', False)}")
|
|
print(f" Last activity: {result.get('last_activity', 'N/A')}")
|
|
|
|
return self.results
|
|
|
|
def save_raw_results(self, output_file: str | None = None):
|
|
"""Save raw results to JSON file."""
|
|
if not output_file:
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
output_file = f"whatsapp_raw_{self.country_code}{self.phone_base}_{timestamp}.json"
|
|
|
|
output_path = Path(output_file)
|
|
|
|
# Prepare output data with full raw responses
|
|
output_data = {
|
|
"test_metadata": {
|
|
"phone_base": f"{self.country_code}{self.phone_base}XXXX",
|
|
"country_code": self.country_code,
|
|
"test_date": datetime.now().isoformat(),
|
|
"method": "10_variations_last_digit_replacement",
|
|
"total_tested": len(self.results)
|
|
},
|
|
"raw_results": self.results
|
|
}
|
|
|
|
# Write to file
|
|
with open(output_path, 'w') as f:
|
|
json.dump(output_data, f, indent=2)
|
|
|
|
print(f"\n💾 Raw results saved to: {output_path}")
|
|
return output_path
|
|
|
|
|
|
async def main():
|
|
"""Main function."""
|
|
if len(sys.argv) != 3:
|
|
print("Usage: python whatsapp_discovery_simple.py <phone_base> <country_code>")
|
|
print("\nAvailable phone bases from our enriched profiles:")
|
|
print(" 3162940 - Bas Schreuder")
|
|
print(" 3161489 - Bjorn de Jong")
|
|
print(" 3165313 - Mylène Da Silva")
|
|
print("\nExample: python whatsapp_discovery_simple.py 3162940 31")
|
|
sys.exit(1)
|
|
|
|
phone_base = sys.argv[1]
|
|
country_code = sys.argv[2]
|
|
|
|
# Check for API token
|
|
if not os.getenv("WHATSAPP_BUSINESS_TOKEN"):
|
|
print("❌ ERROR: WHATSAPP_BUSINESS_TOKEN environment variable not set!")
|
|
print(" Please set it before running:")
|
|
print(" export WHATSAPP_BUSINESS_TOKEN='your_token_here'")
|
|
print(" Or add it to your .env file")
|
|
sys.exit(1)
|
|
|
|
tester = SimpleWhatsAppDiscovery(phone_base, country_code)
|
|
results = await tester.run_test()
|
|
|
|
# Save raw results
|
|
output_file = tester.save_raw_results()
|
|
|
|
print(f"\n✅ Test complete! Raw data saved to {output_file}")
|
|
print(" You can now study the raw API responses in detail.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main()) |