mmcghen
8d382e723f
- Added Walmart scraper to scrape product data from Walmart.com, including category pages and product details. - Introduced a stealth browser module to handle bot protection and improve scraping reliability. - Created a SQLite database for tracking product history, price changes, stock events, and user favorites. - Developed a Discord bot for user interaction, allowing location setting and stock checking at local stores. - Implemented a favorites system to manage priority products and categories with custom notification settings. - Added news aggregation module to fetch and analyze Pokemon TCG news from various sources. - Created tools for API discovery and monitoring, including a backend monitor for detecting new products. - Added unit tests for database operations, product filtering, and API endpoints to ensure functionality. - Enhanced existing modules with improved error handling and logging for better maintainability.
337 lines
10 KiB
Python
337 lines
10 KiB
Python
"""
|
|
HAR File Analyzer for Pokemon Center API Discovery
|
|
|
|
Analyzes a HAR (HTTP Archive) file exported from Chrome DevTools
|
|
to find API endpoints used by Pokemon Center.
|
|
|
|
Usage:
|
|
1. Export HAR from Chrome DevTools Network tab
|
|
2. Save as pokemon_network.har in this folder
|
|
3. Run: python analyze_har.py
|
|
|
|
Or specify a different file:
|
|
python analyze_har.py my_capture.har
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
import re
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse, parse_qs
|
|
from collections import defaultdict
|
|
|
|
# Patterns that indicate interesting API endpoints
|
|
INTERESTING_PATTERNS = [
|
|
r'/api/',
|
|
r'/graphql',
|
|
r'/v\d+/',
|
|
r'product',
|
|
r'catalog',
|
|
r'search',
|
|
r'inventory',
|
|
r'stock',
|
|
r'\.json$',
|
|
r'algolia',
|
|
r'contentful',
|
|
r'commercetools',
|
|
r'demandware',
|
|
r'sfcc',
|
|
]
|
|
|
|
# Skip these resource types
|
|
SKIP_TYPES = [
|
|
'image', 'stylesheet', 'font', 'script', 'media',
|
|
'.png', '.jpg', '.jpeg', '.gif', '.svg', '.ico',
|
|
'.css', '.woff', '.woff2', '.ttf',
|
|
'.js', # Usually not APIs, but we'll catch fetch calls
|
|
]
|
|
|
|
def should_skip(url, mime_type=''):
|
|
"""Check if we should skip this request"""
|
|
url_lower = url.lower()
|
|
mime_lower = mime_type.lower()
|
|
|
|
for skip in SKIP_TYPES:
|
|
if skip in url_lower or skip in mime_lower:
|
|
return True
|
|
return False
|
|
|
|
def is_interesting(url):
|
|
"""Check if URL matches patterns we care about"""
|
|
url_lower = url.lower()
|
|
for pattern in INTERESTING_PATTERNS:
|
|
if re.search(pattern, url_lower):
|
|
return True
|
|
return False
|
|
|
|
def extract_json_preview(content, max_length=500):
|
|
"""Try to extract and preview JSON content"""
|
|
if not content:
|
|
return None
|
|
|
|
try:
|
|
# HAR stores content as text or base64
|
|
text = content.get('text', '')
|
|
if not text:
|
|
return None
|
|
|
|
# Try to parse as JSON
|
|
data = json.loads(text)
|
|
|
|
# Return a preview
|
|
preview = json.dumps(data, indent=2)
|
|
if len(preview) > max_length:
|
|
return preview[:max_length] + "\n... (truncated)"
|
|
return preview
|
|
except:
|
|
return None
|
|
|
|
def analyze_har(har_path):
|
|
"""Analyze a HAR file for API endpoints"""
|
|
|
|
print(f"\nLoading {har_path}...")
|
|
|
|
with open(har_path, 'r', encoding='utf-8') as f:
|
|
har_data = json.load(f)
|
|
|
|
entries = har_data.get('log', {}).get('entries', [])
|
|
print(f"Found {len(entries)} network requests")
|
|
|
|
# Categorize requests
|
|
api_calls = []
|
|
json_responses = []
|
|
graphql_calls = []
|
|
third_party_apis = []
|
|
|
|
domains = defaultdict(int)
|
|
|
|
for entry in entries:
|
|
request = entry.get('request', {})
|
|
response = entry.get('response', {})
|
|
|
|
url = request.get('url', '')
|
|
method = request.get('method', 'GET')
|
|
status = response.get('status', 0)
|
|
mime_type = response.get('content', {}).get('mimeType', '')
|
|
|
|
# Track domains
|
|
parsed = urlparse(url)
|
|
domains[parsed.netloc] += 1
|
|
|
|
# Skip resources we don't care about
|
|
if should_skip(url, mime_type):
|
|
continue
|
|
|
|
# Check for JSON responses
|
|
if 'json' in mime_type.lower():
|
|
content_preview = extract_json_preview(response.get('content', {}))
|
|
json_responses.append({
|
|
'url': url,
|
|
'method': method,
|
|
'status': status,
|
|
'preview': content_preview,
|
|
'headers': {h['name']: h['value'] for h in request.get('headers', [])},
|
|
'response_headers': {h['name']: h['value'] for h in response.get('headers', [])}
|
|
})
|
|
|
|
# Check for GraphQL
|
|
if 'graphql' in url.lower():
|
|
post_data = request.get('postData', {})
|
|
graphql_calls.append({
|
|
'url': url,
|
|
'method': method,
|
|
'body': post_data.get('text', ''),
|
|
'status': status
|
|
})
|
|
|
|
# Check for interesting patterns
|
|
if is_interesting(url):
|
|
api_calls.append({
|
|
'url': url,
|
|
'method': method,
|
|
'status': status,
|
|
'mime': mime_type
|
|
})
|
|
|
|
# Check for third-party APIs
|
|
if any(tp in url.lower() for tp in ['algolia', 'contentful', 'commercetools', 'sfcc']):
|
|
third_party_apis.append({
|
|
'url': url,
|
|
'method': method,
|
|
'status': status
|
|
})
|
|
|
|
# Print analysis
|
|
print("\n" + "=" * 70)
|
|
print("HAR ANALYSIS RESULTS")
|
|
print("=" * 70)
|
|
|
|
# Domains summary
|
|
print("\n[DOMAINS CONTACTED]")
|
|
print("-" * 40)
|
|
for domain, count in sorted(domains.items(), key=lambda x: -x[1])[:15]:
|
|
print(f" {count:4d} requests {domain}")
|
|
|
|
# GraphQL calls
|
|
if graphql_calls:
|
|
print("\n[GRAPHQL ENDPOINTS]")
|
|
print("-" * 40)
|
|
for call in graphql_calls:
|
|
print(f"\n [{call['method']}] {call['url']}")
|
|
print(f" Status: {call['status']}")
|
|
if call['body']:
|
|
try:
|
|
body = json.loads(call['body'])
|
|
if 'query' in body:
|
|
query_preview = body['query'][:200].replace('\n', ' ')
|
|
print(f" Query: {query_preview}...")
|
|
if 'operationName' in body:
|
|
print(f" Operation: {body['operationName']}")
|
|
except:
|
|
print(f" Body: {call['body'][:200]}...")
|
|
|
|
# JSON responses (potential APIs)
|
|
if json_responses:
|
|
print("\n[JSON RESPONSES - Potential APIs]")
|
|
print("-" * 40)
|
|
|
|
# Filter to most interesting ones
|
|
product_related = [r for r in json_responses if 'product' in r['url'].lower()]
|
|
other_json = [r for r in json_responses if 'product' not in r['url'].lower()][:10]
|
|
|
|
for resp in product_related + other_json:
|
|
print(f"\n [{resp['method']}] {resp['url'][:100]}")
|
|
print(f" Status: {resp['status']}")
|
|
|
|
if resp['preview']:
|
|
print(f" Preview:")
|
|
for line in resp['preview'].split('\n')[:10]:
|
|
print(f" {line}")
|
|
|
|
# Third-party APIs
|
|
if third_party_apis:
|
|
print("\n[THIRD-PARTY APIs]")
|
|
print("-" * 40)
|
|
for api in third_party_apis:
|
|
print(f" [{api['method']}] {api['url'][:100]}")
|
|
|
|
# Interesting endpoints
|
|
if api_calls:
|
|
print("\n[OTHER INTERESTING ENDPOINTS]")
|
|
print("-" * 40)
|
|
seen = set()
|
|
for call in api_calls:
|
|
# Dedupe by base URL
|
|
parsed = urlparse(call['url'])
|
|
base = f"{parsed.scheme}://{parsed.netloc}{parsed.path}"
|
|
if base in seen:
|
|
continue
|
|
seen.add(base)
|
|
|
|
print(f" [{call['method']}] {call['url'][:100]}")
|
|
print(f" Status: {call['status']}, Type: {call['mime']}")
|
|
|
|
# Save detailed results
|
|
output_file = Path(har_path).stem + "_analysis.json"
|
|
output_path = Path(har_path).parent / output_file
|
|
|
|
with open(output_path, 'w') as f:
|
|
json.dump({
|
|
'summary': {
|
|
'total_requests': len(entries),
|
|
'json_responses': len(json_responses),
|
|
'graphql_calls': len(graphql_calls),
|
|
'third_party_apis': len(third_party_apis),
|
|
'interesting_endpoints': len(api_calls)
|
|
},
|
|
'domains': dict(domains),
|
|
'json_responses': json_responses,
|
|
'graphql_calls': graphql_calls,
|
|
'third_party_apis': third_party_apis,
|
|
'api_calls': api_calls
|
|
}, f, indent=2)
|
|
|
|
print(f"\n[SAVED] Detailed results saved to: {output_path}")
|
|
|
|
# Recommendations
|
|
print("\n" + "=" * 70)
|
|
print("RECOMMENDATIONS")
|
|
print("=" * 70)
|
|
|
|
if graphql_calls:
|
|
print("""
|
|
[OK] GraphQL endpoint found! This is likely the main data source.
|
|
Next steps:
|
|
1. Examine the query structure in the analysis JSON
|
|
2. Test if the endpoint works without cookies/auth
|
|
3. Build a monitor that polls this endpoint
|
|
""")
|
|
|
|
if any('algolia' in api['url'].lower() for api in third_party_apis):
|
|
print("""
|
|
[OK] Algolia search detected! This is often used for product search.
|
|
Next steps:
|
|
1. Find the Algolia App ID and Search API Key (usually in page source)
|
|
2. Query Algolia directly - very fast and low detection risk
|
|
3. Search for "apiKey" or "applicationId" in the page source
|
|
""")
|
|
|
|
if json_responses and not graphql_calls:
|
|
print("""
|
|
[INFO] JSON responses found but no GraphQL. Check the analysis JSON for:
|
|
- Responses containing product arrays
|
|
- URLs with /api/ or /v1/, /v2/ patterns
|
|
- Look for pagination parameters (page, limit, offset)
|
|
""")
|
|
|
|
if not json_responses and not graphql_calls:
|
|
print("""
|
|
[WARNING] No obvious API endpoints found. Possible reasons:
|
|
- Server-side rendering (data embedded in HTML)
|
|
- API calls blocked by bot protection
|
|
- Need to scroll/interact more to trigger lazy loading
|
|
|
|
Try:
|
|
- Scrolling more on the page before exporting HAR
|
|
- Clicking on product filters
|
|
- Looking at individual product pages
|
|
""")
|
|
|
|
def main():
|
|
# Default HAR file location
|
|
default_har = Path(__file__).parent / "pokemon_network.har"
|
|
|
|
# Check command line args
|
|
if len(sys.argv) > 1:
|
|
har_path = Path(sys.argv[1])
|
|
else:
|
|
har_path = default_har
|
|
|
|
if not har_path.exists():
|
|
print("=" * 70)
|
|
print("HAR File Analyzer")
|
|
print("=" * 70)
|
|
print(f"""
|
|
No HAR file found at: {har_path}
|
|
|
|
To capture a HAR file:
|
|
1. Open Chrome and go to: https://www.pokemoncenter.com/category/tcg-cards
|
|
2. Press F12 to open DevTools
|
|
3. Go to the Network tab
|
|
4. Check "Preserve log" checkbox
|
|
5. Refresh the page (F5)
|
|
6. Scroll down to load more products
|
|
7. Right-click in Network panel → "Save all as HAR with content"
|
|
8. Save as: {default_har}
|
|
9. Run this script again
|
|
|
|
Or specify a HAR file:
|
|
python analyze_har.py path/to/your/file.har
|
|
""")
|
|
return
|
|
|
|
analyze_har(har_path)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|