up/ОАиП/2025-26/36гр/1 сем/gorbunov.py

import socket
import requests
import uuid
import psutil 
import netifaces 
from pythonping import ping as python_ping
import time
import os
import urllib3
from urllib.parse import urlparse

# Отключаем предупреждения SSL для избежания ошибок сертификатов
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

# --- Утилитарные функции ---

def get_size(bytes_count):
    """
    Преобразует байты в удобочитаемый формат (B, KB, MB, GB, TB).
    """
    if not isinstance(bytes_count, (int, float)) or bytes_count is None:
        return "N/A"
        
    for unit in ['B', 'KB', 'MB', 'GB', 'TB', 'PB']:
        if bytes_count < 1024.0:
            return f"{bytes_count:.2f} {unit}"
        bytes_count /= 1024.0
    return f"{bytes_count:.2f} PB"

def format_report(data, target_host):
    """
    Форматирует все собранные данные в одну строку для вывода в файл и консоль.
    """
    timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
    
    report_lines = []
    report_lines.append("="*50)
    report_lines.append("           СВОДНАЯ ИНФОРМАЦИЯ О СЕТИ")
    report_lines.append("="*50)
    report_lines.append(f"Дата и время отчета: {timestamp}")
    report_lines.append("-" * 30)
    
    # Блок 1: Системная информация
    report_lines.append("🖥️  Система:")
    report_lines.append(f"  Имя компьютера: {data.get('hostname', 'N/A')}")
    report_lines.append(f"  Сетевой адаптер: {data.get('interface_name', 'N/A')}")
    report_lines.append(f"  MAC-адрес: {data.get('mac_address', 'N/A')}")
    report_lines.append("-" * 30)
    
    # Блок 2: IP-адреса и шлюз
    report_lines.append(f"🌐 IP-адреса:")
    report_lines.append(f"  Внутренний IP: {data.get('local_ip', 'N/A')}")
    report_lines.append(f"  IP шлюза (роутера): {data.get('gateway_ip', 'N/A')}")
    report_lines.append(f"  Внешний IP: {data.get('external_ip', 'N/A')}")
    report_lines.append("-" * 30)

    # Блок 3: Объем данных
    report_lines.append(f"📊 Общий объем данных (с момента загрузки ОС):")
    report_lines.append(f"  Отправлено: {get_size(data.get('total_sent'))}")
    report_lines.append(f"  Получено: {get_size(data.get('total_recv'))}")
    report_lines.append("-" * 30)
    
    # Блок 4: Пинг
    report_lines.append(f"⏱️  Пинг и потеря пакетов:")
    
    # Пинг до шлюза
    report_lines.append(f"  - До шлюза ({data.get('gateway_ip', 'N/A')}):")
    ping_gw = data.get('ping_gateway_ms')
    loss_gw = data.get('loss_gateway_percent')
    if ping_gw is not None:
        report_lines.append(f"    Средний пинг: {ping_gw:.2f} мс")
        report_lines.append(f"    Потеря пакетов: {loss_gw:.2f}%")
    else:
        report_lines.append("    Не удалось проверить.")
        
    # Пинг до ya.ru
    report_lines.append(f"  - До {target_host}:")
    ping_ext = data.get('ping_ya_ru_ms')
    loss_ext = data.get('loss_ya_ru_percent')
    if ping_ext is not None:
        report_lines.append(f"    Средний пинг: {ping_ext:.2f} мс")
        report_lines.append(f"    Потеря пакетов: {loss_ext:.2f}%")
    else:
        report_lines.append("    Не удалось проверить (проверьте подключение к Интернету).")
    report_lines.append("-" * 30)
    
    # Блок 5: Скорость через прямое скачивание
    report_lines.append(f"📶 Скорость Интернета (прямое скачивание):")
    dl_speed = data.get('download_speed_mbps')
    
    if dl_speed is not None:
        report_lines.append(f"  Скорость загрузки: {dl_speed:.2f} Мбит/с")
        report_lines.append(f"  Использованный сервер: {data.get('speed_test_server', 'N/A')}")
        report_lines.append(f"  Размер тестового файла: {get_size(data.get('test_file_size', 0))}")
        report_lines.append(f"  Время скачивания: {data.get('download_time', 0):.2f} сек")
    else:
        report_lines.append("  Тестирование скорости не было выполнено или завершилось ошибкой.")

    report_lines.append("="*50)
    
    return "\n".join(report_lines)

# --- Функции для измерения ---

def get_network_io_total():
    """Получает общий объем отправленных и полученных данных."""
    try:
        net_io = psutil.net_io_counters()
        return {"total_sent": net_io.bytes_sent, "total_recv": net_io.bytes_recv}
    except Exception:
        return None

def test_download_speed(url, name, test_duration=10):
    """
    Тестирует скорость загрузки с одного сервера в течение заданного времени.
    """
    try:
        print(f"  Тестируем: {name}")
        
        # Настраиваем сессию с таймаутами
        session = requests.Session()
        session.verify = False
        session.timeout = (10, 30)  # (connect timeout, read timeout)
        
        # Начинаем замер времени
        start_time = time.time()
        total_downloaded = 0
        
        # Запрашиваем файл с потоковой передачей
        response = session.get(url, stream=True)
        response.raise_for_status()
        
        # Читаем данные порциями в течение test_duration секунд
        chunk_size = 1024 * 128  # 128 KB chunks
        last_update = start_time
        
        for chunk in response.iter_content(chunk_size=chunk_size):
            if not chunk:
                break
                
            total_downloaded += len(chunk)
            current_time = time.time()
            elapsed = current_time - start_time
            
            # Обновляем прогресс каждую секунду
            if current_time - last_update >= 1:
                speed_mbps = (total_downloaded * 8) / elapsed / (1024 * 1024)
                print(f"    Прогресс: {elapsed:.1f} сек, скорость: {speed_mbps:.2f} Мбит/с, скачано: {get_size(total_downloaded)}")
                last_update = current_time
            
            # Прерываем если достигли целевого времени тестирования
            if elapsed >= test_duration:
                break
        
        download_time = time.time() - start_time
        
        # Проверяем, что тест длился достаточно долго
        if download_time < 1.0:
            print(f"    ⚠ Тест слишком короткий ({download_time:.1f} сек)")
            return None
            
        # Рассчитываем финальную скорость
        speed_bps = (total_downloaded * 8) / download_time  # биты в секунду
        speed_mbps = speed_bps / (1024 * 1024)  # Мбит в секунду
        
        print(f"    ✓ Успешно: {speed_mbps:.2f} Мбит/с за {download_time:.1f} сек, скачано: {get_size(total_downloaded)}")
        
        return {
            "download_speed_mbps": speed_mbps,
            "speed_test_server": name,
            "test_file_size": total_downloaded,
            "download_time": download_time
        }
        
    except requests.exceptions.Timeout:
        print(f"    ✗ Таймаут подключения")
    except requests.exceptions.RequestException as e:
        print(f"    ✗ Ошибка подключения: {str(e)[:80]}")
    except Exception as e:
        print(f"    ✗ Неожиданная ошибка: {str(e)[:80]}")
    
    return None

def download_speed_test(test_duration=10):
    """
    Измеряет скорость загрузки через прямое скачивание файлов с российских серверов.
    Возвращает скорость в Мбит/с.
    """
    print(f"  Тестирование будет длиться {test_duration} секунд...")
    
    # Обновленный список РАБОТАЮЩИХ российских серверов с тестовыми файлами
    # Используем большие файлы для точного тестирования
    test_servers = [
        {
            'name': 'Yandex Disk (архив)',
            'url': 'https://disk.yandex.ru/d/3C3q6FmH_pj2SA/1%20%D0%93%D0%B1%20%D1%82%D0%B5%D1%81%D1%82%D0%BE%D0%B2%D1%8B%D0%B9%20%D1%84%D0%B0%D0%B9%D0%BB.bin?w=1',
            'timeout': 30
        },
        {
            'name': 'Ростелеком Speedtest',
            'url': 'https://speedtest.rt.ru:8080/speedtest/random4000x4000.jpg',
            'timeout': 30
        },
        {
            'name': 'Moscow Data Center',
            'url': 'http://lg.moscow.datacenter.by/speedtest/100MB.bin',
            'timeout': 30
        },
        {
            'name': 'Beeline Speedtest',
            'url': 'http://speedtest.beeline.ru/speedtest/random1000x1000.jpg',
            'timeout': 30
        },
        {
            'name': 'MTS Speedtest',
            'url': 'http://speedtest.mts.ru/speedtest/random4000x4000.jpg',
            'timeout': 30
        },
        {
            'name': 'Rostelecom Backup',
            'url': 'https://mirror.rt.ru/ubuntu-releases/22.04/ubuntu-22.04.3-live-server-amd64.iso',
            'timeout': 30
        },
        {
            'name': 'Yandex Static',
            'url': 'https://yastatic.net/s3/home-static/_/b4/b4bb7395e1b09a5f6aa182b2e2aec097.png',
            'timeout': 20
        },
        {
            'name': 'Mail.ru Static',
            'url': 'https://imgsmail.ru/splash/v25/splash.jpg',
            'timeout': 20
        },
        {
            'name': 'VK Static',
            'url': 'https://vk.com/images/gift/1/512.jpg',
            'timeout': 20
        },
        {
            'name': 'Russian Government',
            'url': 'https://www.gov.ru/static/img/gerb.svg',
            'timeout': 20
        },
        {
            'name': 'Sberbank Static',
            'url': 'https://www.sberbank.ru/portalserver/static/templates/%5BBBHOST%5D/resources/images/logo.svg',
            'timeout': 20
        }
    ]
    
    print(f"  Доступно серверов для тестирования: {len(test_servers)}")
    
    for server in test_servers:
        result = test_download_speed(server['url'], server['name'], test_duration)
        if result:
            return result
        
        # Небольшая пауза между тестами
        time.sleep(1)
    
    # Если все серверы недоступны, пробуем альтернативный метод
    print("  Все основные серверы недоступны, пробуем альтернативные методы...")
    return alternative_speed_test()

def alternative_speed_test():
    """
    Альтернативный метод измерения скорости через многократные запросы.
    """
    try:
        print("  Альтернативный тест: многократные запросы к ya.ru...")
        
        test_url = "https://ya.ru"
        session = requests.Session()
        session.verify = False
        
        total_downloaded = 0
        start_time = time.time()
        iterations = 0
        max_iterations = 20
        target_duration = 8  # Целевая продолжительность теста
        
        while (time.time() - start_time) < target_duration and iterations < max_iterations:
            try:
                response = session.get(test_url, timeout=5)
                total_downloaded += len(response.content)
                iterations += 1
                
                # Небольшая пауза между запросами
                time.sleep(0.2)
                
            except Exception:
                break
        
        download_time = time.time() - start_time
        
        if download_time > 2.0 and total_downloaded > 100 * 1024:
            speed_bps = (total_downloaded * 8) / download_time
            speed_mbps = speed_bps / (1024 * 1024)
            
            print(f"  ✓ Альтернативный тест: {speed_mbps:.2f} Мбит/с за {download_time:.1f} сек, {iterations} запросов")
            
            return {
                "download_speed_mbps": speed_mbps,
                "speed_test_server": "Альтернативный тест (многократные запросы к ya.ru)",
                "test_file_size": total_downloaded,
                "download_time": download_time
            }
            
    except Exception as e:
        print(f"  ✗ Ошибка альтернативного теста: {str(e)[:80]}")
    
    print("  ✗ Все методы измерения скорости не удались")
    return None

def get_ping_and_loss(target, count=8):
    """Измеряет средний пинг и процент потери пакетов."""
    try:
        print(f"  Пинг до {target}...")
        result = python_ping(target, count=count, timeout=3)
        
        # Отображаем прогресс
        if hasattr(result, '_responses'):
            for i, resp in enumerate(result._responses):
                if resp.success:
                    print(f"    Пакет {i+1}: {resp.time_elapsed_ms:.2f} мс")
                else:
                    print(f"    Пакет {i+1}: потерян")
        
        return {"avg_ping_ms": result.rtt_avg_ms, "packet_loss_percent": result.packet_loss}
    except Exception as e:
        print(f"  ✗ Ошибка пинга: {str(e)[:80]}")
        return None

def get_local_net_details():
    """Получает локальный IP, MAC-адрес, имя активного интерфейса и IP шлюза."""
    details = {"local_ip": None, "mac_address": None, "interface_name": None, "gateway_ip": None}
    
    try:
        gws = netifaces.gateways()
        default_route = gws.get('default', {}).get(netifaces.AF_INET)
        
        if default_route:
            details['gateway_ip'] = default_route[0]
            active_interface = default_route[1]
            details['interface_name'] = active_interface

            addrs = netifaces.ifaddresses(active_interface)
            
            if netifaces.AF_INET in addrs:
                details['local_ip'] = addrs[netifaces.AF_INET][0]['addr']
                
            if netifaces.AF_LINK in addrs:
                details['mac_address'] = addrs[netifaces.AF_LINK][0]['addr'].upper().replace('-', ':')

    except Exception as e:
        print(f"  Ошибка получения сетевых деталей: {e}")
    return details

def get_system_info():
    """Собирает имя хоста и внешний IP."""
    info = {"hostname": None, "external_ip": None}

    try:
        info["hostname"] = socket.gethostname()
    except Exception as e:
        print(f"  Ошибка получения hostname: {e}")

    try:
        # Пробуем несколько сервисов для получения внешнего IP
        services = [
            'https://api.ipify.org?format=json',
            'https://ifconfig.me/all.json',
            'http://ip-api.com/json/'
        ]
        
        for service in services:
            try:
                response = requests.get(service, timeout=5, verify=False)
                response.raise_for_status()
                data = response.json()
                
                if 'ip' in data:
                    info["external_ip"] = data['ip']
                    break
                elif 'query' in data:  # ip-api.com format
                    info["external_ip"] = data['query']
                    break
            except:
                continue
                
    except Exception as e:
        print(f"  Ошибка получения внешнего IP: {e}")

    return info

# --- Основной запуск ---

if __name__ == "__main__":
    
    data = {}
    target_host = 'ya.ru'
    output_filename = "network_info_report.txt"
    
    print("="*60)
    print("        ПРОГРАММА ДЛЯ АНАЛИЗА СЕТЕВОГО ПОДКЛЮЧЕНИЯ")
    print("="*60)

    # Сбор данных
    print("\n--- 1. Сбор базовой информации (IP, MAC, Hostname) ---")
    system_info = get_system_info()
    data.update(system_info)
    print(f"  Hostname: {system_info.get('hostname', 'N/A')}")
    print(f"  Внешний IP: {system_info.get('external_ip', 'N/A')}")
    
    local_info = get_local_net_details()
    data.update(local_info)
    print(f"  Локальный IP: {local_info.get('local_ip', 'N/A')}")
    print(f"  MAC-адрес: {local_info.get('mac_address', 'N/A')}")
    print(f"  Интерфейс: {local_info.get('interface_name', 'N/A')}")
    print(f"  Шлюз: {local_info.get('gateway_ip', 'N/A')}")
    
    io_data = get_network_io_total()
    if io_data:
        data.update(io_data)
        print(f"  Отправлено данных: {get_size(io_data.get('total_sent'))}")
        print(f"  Получено данных: {get_size(io_data.get('total_recv'))}")
    
    print("Базовые данные собраны. ✅")

    # Пинги
    if data.get('gateway_ip'):
        print(f"\n--- 2. Проверка пинга до шлюза ({data['gateway_ip']}) ---")
        ping_gw_data = get_ping_and_loss(data['gateway_ip'], count=10)
        if ping_gw_data:
            data["ping_gateway_ms"] = ping_gw_data["avg_ping_ms"]
            data["loss_gateway_percent"] = ping_gw_data["packet_loss_percent"]
            print(f"  Средний пинг: {ping_gw_data['avg_ping_ms']:.2f} мс")
            print(f"  Потеря пакетов: {ping_gw_data['packet_loss_percent']:.2f}%")
        print("Проверка завершена. 📶")

    print(f"\n--- 3. Проверка пинга до {target_host} ---")
    ping_ext_data = get_ping_and_loss(target=target_host, count=10)
    if ping_ext_data:
        data["ping_ya_ru_ms"] = ping_ext_data["avg_ping_ms"]
        data["loss_ya_ru_percent"] = ping_ext_data["packet_loss_percent"]
        print(f"  Средний пинг: {ping_ext_data['avg_ping_ms']:.2f} мс")
        print(f"  Потеря пакетов: {ping_ext_data['packet_loss_percent']:.2f}%")
    print("Проверка завершена. 🎯")
    
    # Speedtest через прямое скачивание
    print("\n--- 4. Проверка скорости сети (прямое скачивание) ---")
    print("Тестирование скорости загрузки с российских серверов...")
    print("⚠ Тест займет около 10 секунд для точного измерения...")
    
    # Тестируем скорость в течение 10 секунд для точности
    speed_data = download_speed_test(test_duration=10)
    if speed_data:
        data.update(speed_data)
        print(f"\n  ИТОГ скорости: {speed_data['download_speed_mbps']:.2f} Мбит/с")
        print("Тестирование скорости завершено. 🚀")
    else:
        print("Тестирование скорости не выполнено (все серверы недоступны).")

    # Форматирование и вывод отчета
    report_content = format_report(data, target_host)
    
    print("\n" + "="*60)
    print("                     ОТЧЕТ")
    print("="*60)
    print(report_content)
    
    try:
        with open(output_filename, 'w', encoding='utf-8') as f:
            f.write(report_content)
        print(f"\n✅ Отчет успешно сохранен в файле: {output_filename}")
        print(f"📂 Путь: {os.path.abspath(output_filename)}")
    except Exception as e:
        print(f"\n❌ Ошибка при записи отчета в файл: {e}")
    
    print("\n" + "="*60)
    print("           ТЕСТИРОВАНИЕ ЗАВЕРШЕНО")
    print("="*60)