montana/Русский/Логистика/mt_api_probe.py

#!/usr/bin/env python3
"""
Probe MT's actual API endpoints by logging in via Playwright and intercepting network calls.
"""
import os, sys, json, time, math, struct, hmac, hashlib, base64
os.chdir(os.path.dirname(os.path.abspath(__file__)))

EMAIL = "operation@mrlogisticcorp.com"
PASSWORD = "NKh9i8Z!7fU9jfi"
TOTP_SECRET = "MNWTEPTFJZBUC32GJFEWY6LVKQ2GGYKH"


def totp(secret):
    s = secret.upper().replace(' ', '')
    pad = (-len(s)) % 8  # correct padding: 0, 2, 4, or 6 chars
    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)


from playwright.sync_api import sync_playwright

api_responses = []

with sync_playwright() as p:
    browser = p.chromium.launch(
        headless=True,
        args=['--no-sandbox', '--disable-blink-features=AutomationControlled']
    )
    context = browser.new_context(
        viewport={'width': 1280, 'height': 900},
        user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 '
                   '(KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36',
    )
    page = context.new_page()

    # Use page.route() to intercept tile API calls and capture their bodies
    tile_responses = []

    def handle_tile_route(route):
        response = route.fetch()
        try:
            body = response.body()
            body_text = body.decode('utf-8', errors='replace')
            tile_responses.append({
                'url': route.request.url,
                'status': response.status,
                'size': len(body),
                'body': body_text[:1000],
            })
        except Exception:
            pass
        route.fulfill(response=response)

    # Intercept tile data requests
    page.route('**/getData/get_data_json_4/**', handle_tile_route)
    page.route('**/getData/**', handle_tile_route)

    # Also store all response URLs for inspection
    intercepted_urls = []
    def on_resp(resp):
        url = resp.url
        skip = ['.js', '.css', '.png', '.jpg', '.gif', '.svg', '.woff', '.ttf',
                'google-analytics', 'hotjar', 'facebook', 'segment.io', 'amplitude']
        if any(s in url for s in skip):
            return
        intercepted_urls.append({'url': url, 'status': resp.status})
    page.on('response', on_resp)

    # === STEP 1: LOGIN ===
    print("Step 1: Login to MT...")
    page.goto('https://www.marinetraffic.com/en/users/login',
              wait_until='domcontentloaded', timeout=30000)
    time.sleep(3)
    print(f"  Redirected to: {page.url}")

    # Email step
    page.fill('input[name="username"]', EMAIL)
    page.click('button[type="submit"]')
    time.sleep(3)
    print(f"  After email: {page.url}")

    # Password step
    page.fill('input[type="password"]', PASSWORD)
    page.click('button[type="submit"]')
    time.sleep(4)
    print(f"  After password: {page.url}")

    # 2FA step
    if 'mfa-login-options' in page.url:
        print("  2FA options screen — clicking Google Authenticator...")
        page.click('button:has-text("Google Authenticator")')
        time.sleep(3)
        otp = totp(TOTP_SECRET)
        print(f"  TOTP code: {otp}")
        page.fill('input[name="code"]', otp)
        page.click('button[type="submit"]')
        time.sleep(5)
        print(f"  After 2FA: {page.url}")

    # Verify login
    logged_in = 'marinetraffic.com' in page.url and 'auth.kpler' not in page.url
    print(f"  Logged in: {logged_in} | URL: {page.url}")
    page.screenshot(path='mt_after_login.png')

    if not logged_in:
        print("  ERROR: Not logged in. Stopping.")
        browser.close()
        sys.exit(1)

    # === STEP 2: LOAD ROTTERDAM MAP ===
    print("\nStep 2: Loading Rotterdam map (zoom 10)...")
    intercepted_urls.clear()
    tile_responses.clear()
    page.goto('https://www.marinetraffic.com/en/ais/home/centerx:4.5/centery:51.9/zoom:10',
              wait_until='load', timeout=30000)
    time.sleep(10)
    page.screenshot(path='mt_rotterdam_map.png')
    print(f"  Page title: {page.title()}")
    print(f"  Intercepted URLs: {len(intercepted_urls)}")
    print(f"  Tile responses captured: {len(tile_responses)}")

    # Show captured tile data
    if tile_responses:
        print("\n=== CAPTURED TILE DATA ===")
        for tr in tile_responses[:5]:
            print(f"  {tr['status']} ({tr['size']}b): {tr['url'][:90]}")
            print(f"    Body: {tr['body'][:300]}")
        with open('mt_tile_data.json', 'w') as f:
            json.dump(tile_responses, f, indent=2)
        print("Saved to mt_tile_data.json")
    else:
        print("\nNO tile responses captured via route interceptor.")

    # === STEP 3: FETCH BODIES OF INTERESTING URLS ===
    print(f"\nStep 3: Fetching bodies of non-static responses...")
    api_responses = []
    for item in intercepted_urls:
        url = item['url']
        status = item['status']
        if status != 200:
            continue
        try:
            # Use page.request (same session/cookies) to re-fetch interesting URLs
            r = page.request.get(url, timeout=10000)
            body = r.body()
            size = len(body)
            if size < 100:
                continue
            body_preview = body[:500].decode('utf-8', errors='replace')
            api_responses.append({'status': status, 'size': size, 'url': url, 'body_preview': body_preview})
            body_lower = body_preview.lower()
            has_vessel = any(k in body_lower for k in ['mmsi', 'shipname', 'shiptype', '"lat"', 'latitude'])
            mark = '*** VESSEL ***' if has_vessel else ''
            print(f"  ({size:7}b) {mark} {url[:110]}")
            if has_vessel:
                print(f"    {body_preview[:200]}")
        except Exception as e:
            pass

    with open('mt_all_responses.json', 'w') as f:
        json.dump(api_responses, f, indent=2)

    # Also try tile API directly via authenticated session
    print("\nStep 4: Direct tile API test (Rotterdam z10)...")
    lat, lon, zoom = 51.9, 4.5, 10
    import math
    lat_rad = math.radians(lat)
    n = 2.0 ** zoom
    x = int((lon + 180.0) / 360.0 * n)
    y = int((1.0 - math.asinh(math.tan(lat_rad)) / math.pi) / 2.0 * n)
    tile_url = f'https://www.marinetraffic.com/getData/get_data_json_4/z:{zoom}/X:{x}/Y:{y}/station:0'
    print(f"  Tile URL: {tile_url}")
    try:
        r = page.request.get(tile_url, headers={
            'Referer': f'https://www.marinetraffic.com/en/ais/home/centerx:{lon}/centery:{lat}/zoom:{zoom}',
            'X-Requested-With': 'XMLHttpRequest',
        })
        body = r.body().decode('utf-8', errors='replace')
        print(f"  Status: {r.status}, Size: {len(body)}")
        print(f"  Response: {body[:300]}")
    except Exception as e:
        print(f"  Error: {e}")

    browser.close()
    print("\nDone!")