Как сделать чтобы я получил информацию о сервере CS2?

Question

[pivodev]

Я хочу получить информацию о сервере кс2 но получаю только информацию о сервере ксго вот строчки кода

def get_server_info(ip, port):
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
            sock.settimeout(5)  # Устанавливаем таймаут на 5 секунд
            sock.connect((ip, port))
            sock.send(b'\xFF\xFF\xFF\xFF\x54Source Engine Query\x00')
            data = sock.recv(4096)
            return data.decode('latin1').split('\x00')[1:]
    except Exception as e:
        logging.error(f"Ошибка при получении информации о сервере: {e}")
        return None

Comments

[Vindicar]

А протокол общения у CS2 и CS:GO одинаковый? Или разный? Если разный, ты уверен, что ты используешь правильный запрос?

[pivodev]

Vindicar,

[Vindicar]

pivodev, ну значит у них протокол общения разный, что не удивительно, игры-то на разных движках. Тогда твой код для CS2 не подойдёт. Я, увы, навскидку в гугле не нашёл описания протокола для CS2. Может, я плохо искал, конечно.

[pivodev]

Vindicar, Спасибо что хоть как-то пытался помочь я это очень ценю ведь мало людей хотят помогать новичкам

Answer 1

[AsQ_QQ]

from ServerGet.ConfigRead import read_all_port, read_ip
import concurrent.futures, socket, a2s
QUERY_TIMEOUT = 4
import socket


class servergetinfo:
    def __init__(self):
        self.ip = read_ip()
        self.ports = read_all_port()


    def get_server_info(self):
        """Функция для получения информации о серверах из конфигурационного файла

        Raises:
            info_except: Информация о серверах

        Returns:
            _type_: Список словарей с информацией о серверах
        """


        info = []
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
                sock.settimeout(5)  # Устанавливаем таймаут на 5 секунд
                for port in self.ports:
                    ip, port = self.ip, int(port)

                    # Запрос подключения к серверу
                    try:
                        socket.getaddrinfo(ip, port)
                    except socket.gaierror:
                        print("Неверный адрес сервера")
                    
                    # Получение информации о сервере
                    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
                        info_future = pool.submit(
                            a2s.info, (ip, port), timeout=QUERY_TIMEOUT)

                    # Инфомарция сервера (временная)
                    info_except = info_future.exception()

                    # Проверка ответа сервера
                    if isinstance(info_except, socket.timeout):
                        print("Сервер не ответил")
                    elif isinstance(info_except, a2s.BrokenMessageError):
                        print("Сервер отправил неверный ответ")
                    elif info_except is not None:
                        raise info_except
                    
                    # Инфомария сервера
                    info_res = info_future.result()
                    info.append(info_res)

                return info
        except Exception as e:
            print(f"Ошибка при получении информации о сервере: {e}")
            return None

Comments

[AsQ_QQ]

from ServerGet import servergetinfo

for info in servergetinfo().get_server_info():
    print(info.server_name)
    break

[AsQ_QQ]

Можно еще так, так будет инфа, айпи, порт)))))

from ServerGet.ConfigRead import read_all_port, read_ip
import concurrent.futures, socket, a2s
QUERY_TIMEOUT = 4
import socket


class servergetinfo:
    def __init__(self):
        self.ip = read_ip()
        self.ports = read_all_port()


    def get_server_info(self):
        """Функция для получения информации о серверах из конфигурационного файла

        Raises:
            info_except: Информация о серверах

        Returns:
            _type_: Список словарей с информацией о серверах
        """


        info = []
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
                sock.settimeout(5)  # Устанавливаем таймаут на 5 секунд
                for port in self.ports:
                    ip, port = self.ip, int(port)

                    # Запрос подключения к серверу
                    try:
                        socket.getaddrinfo(ip, port)
                    except socket.gaierror:
                        print("Неверный адрес сервера")
                    
                    # Получение информации о сервере
                    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
                        info_future = pool.submit(
                            a2s.info, (ip, port), timeout=QUERY_TIMEOUT)

                    # Инфомарция сервера (временная)
                    info_except = info_future.exception()

                    # Проверка ответа сервера
                    if isinstance(info_except, socket.timeout):
                        print("Сервер не ответил")
                    elif isinstance(info_except, a2s.BrokenMessageError):
                        print("Сервер отправил неверный ответ")
                    elif info_except is not None:
                        raise info_except
                    
                    # Инфомария сервера
                    info_res = info_future.result()
                    info.append([info_res, [ip, port]])

                return info
        except Exception as e:
            print(f"Ошибка при получении информации о сервере: {e}")
            return None

Вот запуск:

from ServerGet import servergetinfo

for info in servergetinfo().get_server_info():
    info_server = info[0]
    ip_port = info[1]
    ip, port = ip_port[0], ip_port[1]
    
    print(info_server.map_name)
    print(info_server.player_count)
    print(info_server.max_players)
    print(info_server.server_name)
    print(f"{ip}:{port}")
    break