#!/usr/bin/env python3 """Test: free vessel detail API endpoints — fetch from robots.txt page.""" import asyncio, json, time, struct, hmac, hashlib, base64, sys if hasattr(sys.stdout, 'reconfigure'): sys.stdout.reconfigure(encoding='utf-8', errors='replace', line_buffering=True) EMAIL = "operation@mrlogisticcorp.com" PASSWORD = "NKh9i8Z!7fU9jfi" TOTP_SECRET = "MNWTEPTFJZBUC32GJFEWY6LVKQ2GGYKH" def totp(secret): s = secret.upper().replace(' ', '') pad = (-len(s)) % 8 key = base64.b32decode(s + '=' * pad) counter = int(time.time()) // 30 msg = struct.pack('>Q', counter) h = hmac.new(key, msg, hashlib.sha1).digest() offset = h[-1] & 0x0f code = struct.unpack('>I', h[offset:offset+4])[0] & 0x7fffffff return str(code % 1000000).zfill(6) async def do_login(page, max_retries=3): for attempt in range(max_retries): print(f"LOGIN (attempt {attempt+1}/{max_retries})...", flush=True) try: await page.goto('https://www.marinetraffic.com/en/users/login', wait_until='domcontentloaded', timeout=30000) except Exception as e: print(f" Nav error: {e}", flush=True) continue await asyncio.sleep(3) try: await page.click('button:has-text("AGREE")', timeout=3000) print(" Consent accepted", flush=True) except: pass try: await page.fill('input[name="username"]', EMAIL) await page.click('button[type="submit"]') except Exception as e: print(f" Email error: {e}", flush=True) continue await asyncio.sleep(3) try: await page.fill('input[type="password"]', PASSWORD) await page.click('button[type="submit"]') except Exception as e: print(f" Password error: {e}", flush=True) continue await asyncio.sleep(4) if 'mfa' in page.url.lower() or 'auth.kpler' in page.url: try: await page.click('button:has-text("Google Authenticator")', timeout=3000) await asyncio.sleep(2) except: pass # Wait for fresh TOTP window elapsed = int(time.time()) % 30 remaining = 30 - elapsed if remaining < 8: wait = 30 - elapsed + 2 print(f" TOTP: waiting {wait}s for fresh window...", flush=True) await asyncio.sleep(wait) otp = totp(TOTP_SECRET) print(f" TOTP: {otp}", flush=True) filled = False for selector in ['input[name="code"]', 'input[type="tel"]', 'input[inputmode="numeric"]']: try: await page.fill(selector, otp, timeout=3000) filled = True break except: continue if not filled: # Try any visible text input try: inputs = page.locator('input:visible') count = await inputs.count() for i in range(count): inp = inputs.nth(i) inp_type = await inp.get_attribute('type') or 'text' if inp_type in ('text', 'tel', 'number'): await inp.fill(otp) filled = True break except: pass if filled: await page.click('button[type="submit"]') await asyncio.sleep(8) ok = 'marinetraffic.com' in page.url and 'auth.kpler' not in page.url if ok: print(f" Login OK: {page.url}", flush=True) return True print(f" Login failed: {page.url[:80]}", flush=True) return False async def main(): from playwright.async_api import async_playwright async with async_playwright() as p: browser = await p.chromium.launch( headless=False, args=['--no-sandbox', '--disable-blink-features=AutomationControlled'] ) ctx = await browser.new_context(viewport={'width': 1440, 'height': 900}) page = await ctx.new_page() if not await do_login(page): print("LOGIN FAILED", flush=True) await browser.close() return # Go to lightweight page await page.goto('https://www.marinetraffic.com/robots.txt', wait_until='load', timeout=15000) await asyncio.sleep(2) # Test 1: fetch /general from robots.txt (no navigation to detail page!) print("\n=== TEST 1: /general endpoint (MMSI, IMO, Flag, LOA) ===", flush=True) sids = ['420176', '729727', '9406497', '755836', '713653'] for sid in sids: gen = await page.evaluate("""async () => { try { const r = await fetch('https://www.marinetraffic.com/en/vessels/""" + sid + """/general', { credentials: 'include', cache: 'no-store', headers: {'Accept':'application/json','X-Requested-With':'XMLHttpRequest'} }); if (r.status !== 200) return {error: 'HTTP '+r.status}; return await r.json(); } catch(e) { return {error: e.message}; } }""") if gen.get('error'): print(f" {sid}: ERROR {gen['error']}", flush=True) else: ct = gen.get('commercial_type') or {} print(f" {sid}: {gen.get('name','?')} | " f"MMSI={gen.get('mmsi','?')} IMO={gen.get('imo','?')} " f"Flag={gen.get('countryCode','?')} LOA={gen.get('length','?')} " f"Beam={gen.get('width','?')} Year={gen.get('yearBuilt','?')} | " f"{gen.get('subtype','?')} ({ct.get('market_name','?')})", flush=True) # Test 2: ownership endpoints print("\n=== TEST 2: Ownership endpoints ===", flush=True) ownership_urls = [ f'https://www.marinetraffic.com/en/vessels/420176/ownership', f'https://www.marinetraffic.com/en/vesselDetails/ownership/shipid:420176', ] for url in ownership_urls: r = await page.evaluate("""async () => { try { const r = await fetch('""" + url + """', { credentials: 'include', cache: 'no-store', headers: {'Accept':'application/json','X-Requested-With':'XMLHttpRequest'} }); const text = await r.text(); return {status: r.status, body: text.substring(0, 500)}; } catch(e) { return {error: e.message}; } }""") status = r.get('status', 'err') body = r.get('body', r.get('error', '?'))[:300] print(f" {url.split('420176')[1]}: {status} | {body}", flush=True) # Test 3: position (draught) print("\n=== TEST 3: /position endpoint (draught) ===", flush=True) pos = await page.evaluate("""async () => { try { const r = await fetch('https://www.marinetraffic.com/en/vessels/420176/position', { credentials: 'include', cache: 'no-store', headers: {'Accept':'application/json','X-Requested-With':'XMLHttpRequest'} }); if (r.status !== 200) return {error: 'HTTP '+r.status}; return await r.json(); } catch(e) { return {error: e.message}; } }""") if pos.get('error'): print(f" ERROR: {pos['error']}", flush=True) else: print(f" Draught={pos.get('draught')} Speed={pos.get('speed')} " f"Status={pos.get('navigationalStatus')}", flush=True) # Test 4: speed test — how fast can we fetch /general? print("\n=== TEST 4: Speed test (10 fetches) ===", flush=True) t0 = time.time() for sid in ['420176', '729727', '9406497', '755836', '713653', '738224', '272880', '4610860', '4628227', '7377978']: await page.evaluate("""async () => { const r = await fetch('https://www.marinetraffic.com/en/vessels/""" + sid + """/general', { credentials: 'include', cache: 'no-store', headers: {'Accept':'application/json','X-Requested-With':'XMLHttpRequest'} }); return await r.json(); }""") elapsed = time.time() - t0 print(f" 10 fetches in {elapsed:.1f}s ({elapsed/10:.2f}s/vessel)", flush=True) await browser.close() asyncio.run(main())