nordabiz/scripts/ovh_vps_monitor.py

#!/usr/bin/env python3
"""
OVH VPS Availability Monitor for Warsaw (WAW)
Sprawdza dostępność VPS-1, VPS-2, VPS-3, VPS-4 w Warszawie.
Powiadamia przez macOS notification + opcjonalnie Telegram.

Użycie:
    python3 scripts/ovh_vps_monitor.py              # jednorazowe sprawdzenie
    python3 scripts/ovh_vps_monitor.py --daemon      # co 10 minut

Konfiguracja OVH API (jednorazowo):
    1. Wejdź na https://eu.api.ovh.com/createToken/
    2. Zaloguj się kontem pm861830-ovh
    3. Ustaw:
       - GET /order/cart/*
       - POST /order/cart/*
       - DELETE /order/cart/*
    4. Wpisz klucze do ~/.ovh.conf (format poniżej)

~/.ovh.conf:
    [default]
    endpoint=ovh-eu
    application_key=TWÓJ_APP_KEY
    application_secret=TWÓJ_APP_SECRET
    consumer_key=TWÓJ_CONSUMER_KEY
"""

import json
import os
import subprocess
import sys
import time
from datetime import datetime
from pathlib import Path

# --- Configuration ---
MODELS = {
    'vps-2025-model1': 'VPS-1 (4 vCores, 8 GB, 75 GB SSD) — 23,50 PLN/m',
    'vps-2025-model2': 'VPS-2 (6 vCores, 12 GB, 100 GB NVMe) — 36,21 PLN/m',
    'vps-2025-model3': 'VPS-3 (8 vCores, 24 GB, 200 GB NVMe) — 72,42 PLN/m',
    'vps-2025-model4': 'VPS-4 (12 vCores, 48 GB, 300 GB NVMe) — 133,96 PLN/m',
}

TARGET_DC = 'WAW'
CHECK_INTERVAL = 600  # 10 minut
STATE_FILE = Path(__file__).parent / '.ovh_vps_monitor_state.json'
TELEGRAM_BOT_TOKEN = os.environ.get('TELEGRAM_BOT_TOKEN', '')
TELEGRAM_CHAT_ID = os.environ.get('TELEGRAM_CHAT_ID', '')


def check_availability_curl():
    """Sprawdza dostępność VPS w WAW przez publiczne API OVH (cart flow)."""
    import urllib.request
    import urllib.error

    results = {}

    # 1. Utwórz koszyk
    req = urllib.request.Request(
        'https://eu.api.ovh.com/1.0/order/cart',
        data=json.dumps({'ovhSubsidiary': 'PL'}).encode(),
        headers={'Content-Type': 'application/json'},
        method='POST'
    )
    try:
        with urllib.request.urlopen(req, timeout=15) as resp:
            cart = json.loads(resp.read())
            cart_id = cart['cartId']
    except Exception as e:
        print(f'[ERROR] Nie mogę utworzyć koszyka: {e}')
        return {}

    # 2. Dla każdego modelu — dodaj do koszyka i sprawdź DC
    for plan_code, plan_name in MODELS.items():
        try:
            # Dodaj VPS do koszyka
            add_req = urllib.request.Request(
                f'https://eu.api.ovh.com/1.0/order/cart/{cart_id}/vps',
                data=json.dumps({
                    'planCode': plan_code,
                    'duration': 'P1M',
                    'pricingMode': 'default',
                    'quantity': 1
                }).encode(),
                headers={'Content-Type': 'application/json'},
                method='POST'
            )
            with urllib.request.urlopen(add_req, timeout=15) as resp:
                item = json.loads(resp.read())
                item_id = item['itemId']

            # Skonfiguruj WAW
            dc_req = urllib.request.Request(
                f'https://eu.api.ovh.com/1.0/order/cart/{cart_id}/item/{item_id}/configuration',
                data=json.dumps({'label': 'vps_datacenter', 'value': TARGET_DC}).encode(),
                headers={'Content-Type': 'application/json'},
                method='POST'
            )
            with urllib.request.urlopen(dc_req, timeout=15) as resp:
                dc_result = json.loads(resp.read())

            # Skonfiguruj OS i region
            for cfg in [
                {'label': 'vps_os', 'value': 'Ubuntu 24.04'},
                {'label': 'region', 'value': 'europe'},
            ]:
                cfg_req = urllib.request.Request(
                    f'https://eu.api.ovh.com/1.0/order/cart/{cart_id}/item/{item_id}/configuration',
                    data=json.dumps(cfg).encode(),
                    headers={'Content-Type': 'application/json'},
                    method='POST'
                )
                with urllib.request.urlopen(cfg_req, timeout=15) as resp:
                    pass

            # Sprawdź podsumowanie koszyka (publiczne, nie wymaga auth)
            summary_req = urllib.request.Request(
                f'https://eu.api.ovh.com/1.0/order/cart/{cart_id}/summary',
                method='GET'
            )
            try:
                with urllib.request.urlopen(summary_req, timeout=15) as resp:
                    summary = json.loads(resp.read())
                    results[plan_code] = 'available'
            except urllib.error.HTTPError as e:
                error_body = e.read().decode() if e.fp else ''
                if 'stock' in error_body.lower() or 'unavailable' in error_body.lower():
                    results[plan_code] = 'out_of_stock'
                elif e.code == 401:
                    # Auth wymagane — nie wiemy na pewno, spróbuj auth flow
                    results[plan_code] = 'needs_auth'
                else:
                    results[plan_code] = 'unknown'

            # Usuń item z koszyka
            del_req = urllib.request.Request(
                f'https://eu.api.ovh.com/1.0/order/cart/{cart_id}/item/{item_id}',
                method='DELETE'
            )
            try:
                with urllib.request.urlopen(del_req, timeout=15) as resp:
                    pass
            except Exception:
                pass

        except urllib.error.HTTPError as e:
            error_body = e.read().decode() if e.fp else ''
            if 'stock' in error_body.lower() or 'not available' in error_body.lower():
                results[plan_code] = 'out_of_stock'
            else:
                results[plan_code] = f'error: {e.code}'
        except Exception as e:
            results[plan_code] = f'error: {e}'

    # Usuń koszyk
    try:
        del_cart = urllib.request.Request(
            f'https://eu.api.ovh.com/1.0/order/cart/{cart_id}',
            method='DELETE'
        )
        urllib.request.urlopen(del_cart, timeout=15)
    except Exception:
        pass

    return results


def check_availability_ovh_lib():
    """Sprawdza dostępność VPS w WAW przez ovh Python library (z autoryzacją)."""
    try:
        import ovh
    except ImportError:
        print('[WARN] Brak biblioteki ovh — pip3 install ovh')
        return {}

    conf_path = Path.home() / '.ovh.conf'
    if not conf_path.exists():
        print(f'[WARN] Brak {conf_path} — użyj trybu bez autoryzacji')
        return {}

    try:
        client = ovh.Client()
    except Exception as e:
        print(f'[ERROR] Nie mogę połączyć z OVH API: {e}')
        return {}

    results = {}

    for plan_code, plan_name in MODELS.items():
        try:
            # Utwórz koszyk
            cart = client.post('/order/cart', ovhSubsidiary='PL')
            cart_id = cart['cartId']
            client.post(f'/order/cart/{cart_id}/assign')

            # Dodaj VPS
            item = client.post(f'/order/cart/{cart_id}/vps',
                               planCode=plan_code, duration='P1M',
                               pricingMode='default', quantity=1)
            item_id = item['itemId']

            # Skonfiguruj WAW + OS + region
            client.post(f'/order/cart/{cart_id}/item/{item_id}/configuration',
                        label='vps_datacenter', value=TARGET_DC)
            client.post(f'/order/cart/{cart_id}/item/{item_id}/configuration',
                        label='vps_os', value='Ubuntu 24.04')
            client.post(f'/order/cart/{cart_id}/item/{item_id}/configuration',
                        label='region', value='europe')

            # Validate checkout (GET = validate, POST = place order)
            checkout = client.get(f'/order/cart/{cart_id}/checkout')
            # Jeśli doszliśmy tutaj — VPS jest dostępny w WAW!
            results[plan_code] = 'available'

            # Cleanup
            client.delete(f'/order/cart/{cart_id}')

        except Exception as e:
            error_msg = str(e).lower()
            if 'stock' in error_msg or 'not available' in error_msg or 'unavailable' in error_msg:
                results[plan_code] = 'out_of_stock'
            elif 'expired' in error_msg:
                results[plan_code] = 'error_expired'
            else:
                results[plan_code] = f'error: {e}'

            # Cleanup
            try:
                client.delete(f'/order/cart/{cart_id}')
            except Exception:
                pass

    return results


def notify_macos(title, message):
    """Powiadomienie macOS."""
    subprocess.run([
        'osascript', '-e',
        f'display notification "{message}" with title "{title}" sound name "Glass"'
    ], check=False)


def notify_telegram(message):
    """Powiadomienie Telegram."""
    if not TELEGRAM_BOT_TOKEN or not TELEGRAM_CHAT_ID:
        return
    import urllib.request
    url = f'https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage'
    data = json.dumps({
        'chat_id': TELEGRAM_CHAT_ID,
        'text': message,
        'parse_mode': 'Markdown'
    }).encode()
    req = urllib.request.Request(url, data=data,
                                headers={'Content-Type': 'application/json'})
    try:
        urllib.request.urlopen(req, timeout=10)
    except Exception as e:
        print(f'[WARN] Telegram notification failed: {e}')


def load_state():
    """Wczytaj poprzedni stan."""
    if STATE_FILE.exists():
        return json.loads(STATE_FILE.read_text())
    return {}


def save_state(state):
    """Zapisz stan."""
    STATE_FILE.write_text(json.dumps(state, indent=2))


def run_check():
    """Główna funkcja sprawdzająca."""
    now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    print(f'\n[{now}] Sprawdzam dostępność VPS w {TARGET_DC}...')

    # Preferuj auth flow (dokładniejszy), fallback na public API
    conf_path = Path.home() / '.ovh.conf'
    if conf_path.exists():
        print('  Tryb: OVH API (z autoryzacją)')
        results = check_availability_ovh_lib()
    else:
        print('  Tryb: Public API (bez autoryzacji — mniej dokładny)')
        results = check_availability_curl()

    if not results:
        print('  [WARN] Brak wyników — problem z API?')
        return

    prev_state = load_state()
    new_available = []

    for plan_code, status in results.items():
        plan_name = MODELS.get(plan_code, plan_code)
        icon = '✅' if status == 'available' else '❌' if 'stock' in str(status) else '❓'
        print(f'  {icon} {plan_name}: {status}')

        # Czy to nowa dostępność?
        prev = prev_state.get(plan_code, '')
        if status == 'available' and prev != 'available':
            new_available.append(plan_name)

    # Powiadom o nowych dostępnościach
    if new_available:
        msg_lines = ['🟢 VPS dostępny w Warszawie!'] + [f'  • {n}' for n in new_available]
        msg_lines.append(f'\n🔗 https://www.ovhcloud.com/pl/vps/')
        message = '\n'.join(msg_lines)

        print(f'\n  🔔 POWIADOMIENIE: {message}')
        notify_macos('OVH VPS Warszawa!', '\n'.join(new_available))
        notify_telegram(message)
    else:
        print('  Brak nowych dostępności.')

    save_state(results)


def main():
    daemon = '--daemon' in sys.argv

    if daemon:
        print(f'Uruchamiam monitoring VPS w {TARGET_DC} (co {CHECK_INTERVAL}s)...')
        print(f'Stan zapisywany w: {STATE_FILE}')
        print('Ctrl+C aby zatrzymać\n')
        while True:
            try:
                run_check()
                time.sleep(CHECK_INTERVAL)
            except KeyboardInterrupt:
                print('\nZatrzymano.')
                break
    else:
        run_check()


if __name__ == '__main__':
    main()