benchmark_simple.py

import urllib.request, json, time, uuid, os

PYTHON = 'https://opticparse-python-sg.onrender.com'
NODE = 'https://opticparse-1opticparse-node-sg.onrender.com'
SUPABASE_URL = 'https://xxmvhvxeglsjbewlouqg.supabase.co'

# Load Supabase service key from .env in repo root
SERVICE_KEY = ''
env_path = os.path.join(os.path.dirname(__file__), '.env')
with open(env_path, encoding='utf-8') as f:
    for line in f:
        if line.startswith('SUPABASE_SERVICE_KEY='):
            SERVICE_KEY = line.strip().split('=', 1)[1]

# Create test user
req = urllib.request.Request(
    f'{SUPABASE_URL}/auth/v1/admin/users',
    method='POST',
    headers={'Content-Type': 'application/json', 'apikey': SERVICE_KEY, 'Authorization': f'Bearer {SERVICE_KEY}'},
    data=json.dumps({'email': f'bench_{uuid.uuid4().hex[:6]}@test.com', 'password': 'Test123!', 'email_confirm': True}).encode()
)
user_id = json.loads(urllib.request.urlopen(req).read())['id']

# Generate OpticParse API key
req2 = urllib.request.Request(
    f'{PYTHON}/gateway/keys/generate',
    method='POST',
    headers={'Content-Type': 'application/json'},
    data=json.dumps({'user_id': user_id}).encode()
)
api_key = json.loads(urllib.request.urlopen(req2, timeout=30).read())['api_key']
print('API key (truncated):', api_key[:20] + '...')

def scrape(url, query, schema, name):
    start = time.time()
    try:
        req = urllib.request.Request(
            f'{PYTHON}/api/vision-scrape',
            method='POST',
            headers={'Content-Type': 'application/json', 'X-API-Key': api_key},
            data=json.dumps({'target_url': url, 'extraction_query': query, 'response_schema': schema}).encode()
        )
        res = urllib.request.urlopen(req, timeout=90)
        _ = json.loads(res.read())
        elapsed = time.time() - start
        print('PASS', name, f'({elapsed:.1f}s)')
        return elapsed, True
    except Exception as e:
        elapsed = time.time() - start
        print('FAIL', name, f'({elapsed:.1f}s)', str(e)[:80])
        return elapsed, False

def phish(url, name, expected='safe'):
    start = time.time()
    try:
        req = urllib.request.Request(
            f'{NODE}/api/phish-detect',
            method='POST',
            headers={'Content-Type': 'application/json'},
            data=json.dumps({'url': url}).encode()
        )
        res = urllib.request.urlopen(req, timeout=90)
        body = json.loads(res.read())
        elapsed = time.time() - start
        verdict = body.get('verdict', '?')
        confidence = body.get('confidence_score_percentage', 0)
        cached = body.get('cached', False)
        correct = verdict == expected
        print(name, verdict, f'({confidence}%)', 'CACHED' if cached else 'FRESH', f'({elapsed:.1f}s)')
        return elapsed, correct
    except Exception as e:
        elapsed = time.time() - start
        print('FAIL', name, f'({elapsed:.1f}s)', str(e)[:60])
        return elapsed, False

print('\n=== OPTICPARSE BENCHMARK ===')
opt_cases = [
    ('https://news.ycombinator.com', 'HN', {'posts': [{'title': 'string', 'points': 'number'}]}),
    ('https://github.com/trending', 'GitHub', {'repos': [{'name': 'string', 'stars': 'string'}]}),
    ('https://vercel.com/pricing', 'Vercel', {'plans': [{'name': 'string', 'price': 'string'}]}),
    ('https://www.cloudflare.com/plans/', 'Cloudflare', {'plans': [{'name': 'string', 'price': 'string'}]}),
    ('https://stripe.com/pricing', 'Stripe', {'products': [{'name': 'string', 'price': 'string'}]})
]
opt_times, opt_results = [], []
for url, name, schema in opt_cases:
    t, r = scrape(url, f'Top items for {name}', schema, name)
    opt_times.append(t); opt_results.append(r)
# cache test
t, r = scrape('https://news.ycombinator.com', 'HN Cache', {'posts': [{'title': 'string'}]}, 'HN Cache')
opt_times.append(t); opt_results.append(r)
opt_pass = sum(opt_results)
opt_avg = sum(opt_times) / len(opt_times) if opt_times else 0
print(f'OpticParse: {opt_pass}/{len(opt_results)} passed | Avg: {opt_avg:.1f}s')

print('\n=== PHISHVISION BENCHMARK ===')
ph_cases = [
    ('https://google.com', 'Google'),
    ('https://microsoft.com', 'Microsoft'),
    ('https://accounts.google.com', 'Google Login'),
    ('https://github.com/login', 'GitHub Login'),
    ('https://apple.com', 'Apple'),
    ('https://google.com', 'Google Cache')
]
ph_times, ph_results = [], []
for url, name in ph_cases:
    t, r = phish(url, name)
    ph_times.append(t); ph_results.append(r)
ph_pass = sum(ph_results)
ph_fresh = [t for t in ph_times if t >= 2]
ph_cached = [t for t in ph_times if t < 2]
ph_fresh_avg = sum(ph_fresh) / len(ph_fresh) if ph_fresh else 0
ph_cached_avg = sum(ph_cached) / len(ph_cached) if ph_cached else 0
print(f'PhishVision: {ph_pass}/{len(ph_results)} correct')
print(f'Fresh avg: {ph_fresh_avg:.1f}s')
print(f'Cached avg: {ph_cached_avg:.2f}s')

print('\n=== SECURITY TESTS ===')
# Fake key
try:
    urllib.request.urlopen(urllib.request.Request(
        f'{PYTHON}/api/vision-scrape', method='POST',
        headers={'Content-Type': 'application/json', 'X-API-Key': 'fake-key-123'},
        data=json.dumps({'target_url': 'https://example.com', 'extraction_query': 'test'}).encode()
    ), timeout=10)
    print('Fake key NOT rejected')
except urllib.error.HTTPError as e:
    print('Fake key rejected', e.code)
# Internal URL
try:
    urllib.request.urlopen(urllib.request.Request(
        f'{NODE}/api/phish-detect', method='POST',
        headers={'Content-Type': 'application/json'},
        data=json.dumps({'url': 'http://127.0.0.1/secret'}).encode()
    ), timeout=10)
    print('Internal URL NOT blocked')
except urllib.error.HTTPError as e:
    print('Internal URL blocked', e.code)