[ACS] функция RCON + модуль для GameCMS (Python)

Сообщения
403
Реакции
114
Помог
2 раз(а)
Я тут подумал, что отпуск по-тихонечку подходит к концу и, возможно, многое из того, что я написал за 2 месяца просто пропадет в недрах моих ноутбуков навсегда... Это было интересно и познавательно, но возможно кому-то пригодится. Полнофункциональный вариант бота и необходимые скрипты управления HLDS+HLTV выложу чуть позже, после проведения косметических процедур...

Код:
##################################
# (c) 2023, ACS RedFoxxx Edition #
##################################

def acs_rcon_cmd(addr, acs_global_password, cmd):
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
        message = b'\xff\xff\xff\xffchallenge rcon'
        s.settimeout(1);
        s.sendto(message, addr)
        try:
            data, acs_global_address = s.recvfrom(1024)
        except Exception as err:
            return acs_err_no_response_from % addr[0]+':'+str(addr[1])
        if data[0:13] == b'\xff\xff\xff\xffchallenge':
            challenge = data[19:-2]
            message = b'\xff\xff\xff\xffrcon ' + challenge + b' ' + acs_global_password.encode() + b' ' + cmd.encode() + b'\n'
            s.sendto(message, addr)
            buffer = b''
            while True:
                    try:
                        data, acs_global_address = s.recvfrom(4096)
                    except Exception:
                        break;
                    if len(data) > 0:
                        if data[0:4] == b'\xfe\xff\xff\xff':
                            data = data[9:]
                        if data[0:4] == b'\xff\xff\xff\xff':
                            data = data[4:]
                        if data[0] == 0x6c:
                            data = data[1:-2]
                        buffer += data
                    else:
                        break

            return buffer.decode().rstrip('\n')
        else:
            return acs_err_challenge_not_aviable % data
Код:
#!/usr/bin/python3

##################################
# (c) 2023, ACS RedFoxxx Edition #
##################################

import socket, MySQLdb, MySQLdb.cursors, re
import asyncio
from ctypes import *
from pprint import pprint

def acs_build_arr(arr, key_fields, row, is_loop=False):
    tmp_arr={}
    _k1 = _k2 = -1
    if type(arr) is list:
        if len(arr) == 0:
            arr.append({})
            arr = arr[-1]
        else:
            _t = arr[-1]
            _k1 = _t[list(_t)[0]]
            _k2 = row[list(_t)[0]]
            if _k1 != _k2:
                arr.append({})
                arr = arr[-1]
    for i, val in key_fields.items():
        if val != 0:
            if i not in arr or is_loop:
                if type(arr) is list:
                    arr = arr[-1][i]
                    _t = arr[-1]
                    _k3 = _t[list(_t)[0]]
                    _k4 = row[list(_t)[0]]
                    if _k3 != _k4:
                        arr.append({})
                        arr = arr[-1]
                        tmp_arr = acs_build_arr(arr, val, row, True)
                    continue
                else:
                    arr[i] = []
            tmp_arr = acs_build_arr(arr[i], val, row, True)
        else:
            tmp_arr[i] = row[i]
            if i not in arr:
                if _k1 != _k2 and (_k1 > 0 and _k2 > 0) or (_k1 < 0 and _k2 < 0):
                    arr[i] = row[i]
    return tmp_arr

localhost = socket.gethostbyname(socket.gethostname())
print('Local IP is %s' % localhost)

mysql_host = ""
mysql_db   = ""
mysql_user = ""
mysql_pass = ""

acs_tg_user_index   = "telegram"
acs_global_servers  = list()
acs_global_services = {}
acs_global_users    = {}

def acs_check_database_flags(id, flags="", is_strict_check=False):
    result = is_strict_check
    if flags:
        if id in acs_global_users:
            for adm in acs_global_users[id]["admins"]:
                for svc in adm["services"]:
                    s_flag = svc["s_rights"].replace("_nick_prefix", "")
                    if s_flag:
                        for f in list(flags):
                            if is_strict_check:
                                if f not in s_flag:
                                    result = False
                                    break
                            else:
                                result |= (f in s_flag)
                                if result:
                                    break
        else:
            result = False
    else:
        result = False
    return result

def acs_copy_dict_list(lst, data):
    for row in data:
        lst.append({})
        _arr=lst[-1]
        for key in row:
            _arr[key] = row[key]

def acs_read_database(servers, services, users):
    print("Connection to the Database...")
    try:
        db = MySQLdb.connect(host = mysql_host,    # your host, usually localhost
                             user = mysql_user,    # your username
                             passwd = mysql_pass,  # your password
                             db = mysql_db,        # name of the data base
                             charset = "utf8",
                             init_command = "SET NAMES UTF8",
                             cursorclass = MySQLdb.cursors.DictCursor)
    except Exception as err:
        print("Connected error with %s" % err)
        return False
    cur = db.cursor()
    cur.execute("select name, ip, port, rcon_password from servers where rcon_password != ''")
    acs_copy_dict_list(servers, cur.fetchall())
    if (not servers):
        print('Error! Servers not found, check MySQL parametrs')
        return False
    print('Finded %d server(s)' % len(servers))
    cur.execute("select name, rights from services")
    rows=cur.fetchall()
    if not rows:
        print('Error! Services not found, check MySQL parametrs')
        #exit(-1)
    for row in rows:
        acs_global_services[row['rights']] = row['name']
    #pprint(acs_global_services)
    #exit(0)
    #acs_copy_dict_list(services, cur.fetchall())
    if (not services):
        print('Error! Services not found, check MySQL parametrs')
        return False
    print("Found %s service(s)" % len(acs_global_services))
    #pprint(acs_global_services)
    sql="select u.id as id, u.name, u.nick, u.email, u.vk, u.telegram, u.steam_id, u.prefix, a.id as a_id, a.name as a_name, a.pass as a_pass, a.type as a_type, a.server as a_server, a.active as a_active, a_svc.id as s_id, a_svc.service as s_service, a_svc.service_time as s_service_time, a_svc.bought_date as s_bought_date, a_svc.ending_date as s_ending_time, svc.name as s_name, svc.rights as s_rights, svc.server as s_server from users as u, admins as a, admins__services as a_svc, services as svc where a.user_id = u.id and a_svc.admin_id = a.id and a_svc.service = svc.id and u.telegram <> '' order by u.id, a.id, svc.id"
    #print("SQL = %s" % sql )
    cur.execute(sql)
    user_key_fields = {
        "id" : 0, "name" : 0, "nick" : 0, "email" : 0, "vk" : 0, "telegram" : 0, "steam_id" : 0, "prefix" : 0,
        "admins" : {
            "a_id" : 0, "a_name" : 0, "a_pass" : 0, "a_type" : 0, "a_server" : 0, "a_active" : 0,
            "services" : {
                "s_id" : 0, "s_service" : 0, "s_service_time" : 0, "s_bought_date" : 0, "s_ending_time" : 0, "s_name" : 0, "s_rights" : 0, "s_server" : 0
            }
        }
    }
    rows = cur.fetchall()
    if (not rows):
        print("Error! Users with TG not found, check MySQL parametrs")
        return False
    print("Found %s user(s)" % len(rows))
    for row in rows:
        _index=row[acs_tg_user_index]
        if _index == '' or not _index.isnumeric():
            continue
        if _index not in users:
            users[_index] = {}
        acs_build_arr(users[_index], user_key_fields, row)
    db.close()
    return True
 
Последнее редактирование:
Сообщения
365
Реакции
9
Помог
5 раз(а)
Обратите внимание, если вы хотите заключить сделку с этим пользователем, он заблокирован
ArKaNeMaN, а зачем он меня вставил ? Не понятно🤔 это создаёт дискуссию и дезинформацию, если ради шутки то ладно

Refresh, но вот мне интересно, ты наверное не понял сути того, что я писал статью, и не менял оригинального автора, слегка были внесены изменения исходного кода, я изменил сборку плагина, текстуру небо, красивое и понятное оформление для тех ребята кто не понял, как использовать плагин и прописывать в ini файл, дополнил недочёты, и предоставил так ребятам, как бы и сам использовал для себя, вообщем по этому назвал свой архив со своим именем Edition, в принципи в статье все расписал
 
Сообщения
403
Реакции
114
Помог
2 раз(а)
RedFoxxx, это шутка, я тебя обещал и сдержал слово...:pardon: Если не нравится, я удалю :blush2: Меня не волнуют чужие копирайты от слова совсем. Если я не хочу что-то показать, я просто это делаю так, что никто, никогда, ничего не поймет. Для понимания принципов программирования и работы алгоритмов, в любом языке, нужны примеры. Единственное, против чего я выступаю, это использование результатов чужой интеллектуальной деятельности в коммерческих целях.
 
Последнее редактирование:
Сообщения
365
Реакции
9
Помог
5 раз(а)
Обратите внимание, если вы хотите заключить сделку с этим пользователем, он заблокирован
Refresh, братец все норм:)
 

Пользователи, просматривающие эту тему

Сейчас на форуме нет ни одного пользователя.
Сверху Снизу