#!/usr/bin/env python3
import os
import pprint
import re
import sys
import zlib
import time
import shutil
import struct
import curses
import pprint
import tarfile
import logging
import datetime
import requests
import threading
import subprocess
import yaml
import click
import colorama
from pathlib import Path
from PyQt6 import QtWidgets
from rcon.source import Client
import hlnaui
home_dir = Path.home()
logging.basicConfig(stream=sys.stderr, level=logging.CRITICAL)
def find_file(path):
"""Находим все конфиги в зависимости от пути"""
arr = next(os.walk(path), (None, None, []))[2] # [] if no file
if '.directory' in arr:
arr.remove('.directory')
return arr
dir_config = f"{home_dir}/.config/hlna/"
dir_maps_ark = f"{dir_config}ARK/"
dir_deactivated = f"{dir_maps_ark}deactivated/"
list_config = find_file(dir_maps_ark)
delist_config = find_file(dir_deactivated)
list_allconfigs = list_config + delist_config
def create_dir(directory):
"""Проверка и создание директории"""
if not os.path.exists(directory):
os.makedirs(directory)
create_dir(dir_config)
def path_server():
"""Получение пути для хранения файлов серверов, записываем путь в yaml файл"""
dir_server = input(f"""Укажите путь до каталога, где будут храниться файлы сервера. По-умолчанию {home_dir}/Servers/
:""")
if not dir_server:
dir_server = f"{home_dir}/Servers/"
yaml_create("path_server", dir_server)
return dir_server
@click.group()
def hlna():
pass
def unpack(src, dst):
"""Добавить документацию"""
with open(src, 'rb') as f:
sigver = struct.unpack('q', f.read(8))[0]
unpacked_chunk = f.read(8)
packed = f.read(8)
unpacked = f.read(8)
size_unpacked_chunk = struct.unpack('q', unpacked_chunk)[0]
size_packed = struct.unpack('q', packed)[0]
size_unpacked = struct.unpack('q', unpacked)[0]
# Verify the integrity of the Archive Header
if sigver == 2653586369:
if isinstance(size_unpacked_chunk, int) and isinstance(size_packed, int) and isinstance(size_unpacked, int):
logging.info("Archive is valid.")
logging.debug(
f"Archive header size information. Unpacked Chunk: {size_unpacked_chunk}({unpacked_chunk}) Full Packed: {size_packed}({packed}) Full Unpacked: {size_unpacked}({unpacked})")
# Obtain the Archive Compression Index
compression_index = []
size_indexed = 0
while size_indexed < size_unpacked:
raw_compressed = f.read(8)
raw_uncompressed = f.read(8)
compressed = struct.unpack('q', raw_compressed)[0]
uncompressed = struct.unpack('q', raw_uncompressed)[0]
compression_index.append((compressed, uncompressed))
size_indexed += uncompressed
logging.debug(
f"{len(compression_index)}: {size_indexed}/{size_unpacked} ({compressed}/{uncompressed}) - {raw_compressed} - {raw_uncompressed}")
if size_unpacked != size_indexed:
msg = f"Header-Index mismatch. Header indicates it should only have {size_unpacked} bytes when uncompressed but the index indicates {size_indexed} bytes."
logging.critical(msg)
return print_line(msg, flag="RED")
# Read the actual archive data
data = b''
read_data = 0
for compressed, uncompressed in compression_index:
compressed_data = f.read(compressed)
uncompressed_data = zlib.decompress(compressed_data)
# Verify the size of the data is consistent with the archives index
if len(uncompressed_data) == uncompressed:
data += uncompressed_data
read_data += 1
# Verify there is only one partial chunk
if len(uncompressed_data) != size_unpacked_chunk and read_data != len(compression_index):
msg = f"Index contains more than one partial chunk: was {len(uncompressed_data)} when the full chunk size is {size_unpacked_chunk}, chunk {read_data}/{len(compression_index)}"
logging.critical(msg)
return print_line(msg, flag="RED")
else:
msg = f"Uncompressed chunk size is not the same as in the index: was {len(uncompressed_data)} but should be {uncompressed}."
logging.critical(msg)
return print_line(msg, flag="RED")
else:
msg = f"Data types in the headers should be int's. Size Types: unpacked_chunk({type(size_unpacked_chunk)}), packed({type(size_packed)}), unpacked({type(size_unpacked)})"
logging.critical(msg)
return print_line(msg, flag="RED")
else:
msg = "The signature and format version is incorrect. Signature was {} should be 2653586369.".format(sigver)
logging.critical(msg)
return print_line(msg, flag="RED")
# Write the extracted data to disk
with open(dst, 'wb') as f:
f.write(data)
logging.info("Archive has been extracted.")
def get_external_ip():
response = requests.get('https://api.ipify.org')
return response.text
def print_line(*text, flag="", sep=" ", end="\n"):
"""Добавление обводки вокруг текста, покраска"""
if flag == "RED":
color = colorama.Fore.RED
elif flag == "YELLOW":
color = colorama.Fore.YELLOW
elif flag == "GREEN":
color = colorama.Fore.GREEN
elif flag == "CYAN":
color = colorama.Fore.CYAN
else:
color = colorama.Fore.WHITE
len_text = str(*text)
len_text = len_text.split("\n")
max_string = max(len(str(string)) for string in len_text) + 2
max_length = shutil.get_terminal_size()
max_length = max_length[0]
if max_string > max_length:
len_dots = max_length
else:
len_dots = max_string
print(color + "." * len_dots)
print(color, *text, sep=sep, end=end)
print(color + "." * len_dots + colorama.Style.RESET_ALL)
def zmeyuka(stdscr, func):
print_line(func)
curses.curs_set(0)
stdscr.nodelay(1)
def nachalo_stroki():
while func.do_run:
zmejka = '⠇⠋⠙⠸⠴⠦'
for i in range(len(zmejka)):
print('\r' + zmejka[i], end="", flush=True)
time.sleep(0.25)
th = threading.Thread(target=func)
th.start()
# Запуск потока со змеюкой
zmeyuka_thread = threading.Thread(target=nachalo_stroki)
zmeyuka_thread.start()
# Ожидаем завершение функции
th.join()
# Убираем змеюку
zmeyuka_thread.do_run = False
zmeyuka_thread.join()
curses.curs_set(1)
stdscr.nodelay(0)
def check_int(number=""):
"""Проверка на ввод числа"""
while True:
try:
x = input(number)
if x == "":
return 0
x = int(x)
return x
except ValueError:
print_line("Введите число", flag="CYAN")
@hlna.command(help='Восстановление бэкапов серверов в <hlna restore ark>')
@click.argument('g', nargs=1)
@click.option('-m', default='all', help="Название карты для запуска или all для запуска всех карт")
@click.option('-d', required=True, help="Путь до zip архива")
def restore(g, m, d):
"""Получение пути к файлам внутри архива"""
with tarfile.open(d, 'r') as tar_file:
files = tar_file.getnames()
"""Извлечение файлов"""
for i in files:
with tar_file.extract(d, 'r:gz') as tar_file:
path_extarct = "./" if g == 'test' else "/"
tar_file.extract(i, path_extarct)
print_line(f"i - перемещен", flag="GREEN")
print_line(f"Бэкап {d} восстановлен", flag="GREEN")
@hlna.command(help='Бэкап серверов выбранной игры <hlna backup ark')
@click.argument('g', nargs=1)
@click.option('-m', default='all', help="Название карты для запуска или all для запуска всех карт")
def backup(g, m):
if g == "ark" or g == "ark_test":
name_servers = choose_map(g, m)
config_backup(g, m)
backup_path = f"{dir_server_ark}Backups" if g == "ark" else f"{dir_server_ark}Backups/test_backup"
if not backup_path:
backup_path = f"{dir_server_ark}Backups"
for i in name_servers:
data = read_yaml(g="ark", m=i)
map = data['map']
str(map)
source = [f"{dir_maps_ark}", f"{dir_ark_save}"]
target = f"{backup_path}/{g}_{i}_backup_{time.strftime('%Y_%m_%d')}.tar"
create_dir(backup_path)
with tarfile.open(target, 'w') as mytar:
for source_folder in source:
for root, dirs, files in os.walk(source_folder):
for file in files:
name, ext = os.path.splitext(file)
if ext == '.arkprofile' or name.startswith(map):
path = os.path.join(root, file)
mytar.add(path)
print_line(f"{path} - сохранён", flag="GREEN")
print_line(f"Конфиги сохранены в {target}", flag="GREEN")
elif g == "7days":
pass
else:
print("Поддерживает только ark и 7days")
@hlna.command(help='Выбор игры и сбор настроек для сервера(-ов)')
@click.argument('g', nargs=1)
@click.option('-b', required=False, default=False, help="Конфигурация бэкапов игры")
def config(g, b):
if g == "ark":
if b:
config_backup(g)
else:
config_ark()
elif g == "7days":
config_7daystodie()
else:
print_line("Пока есть только ARK и 7Days xD", flag="YELLOW")
def ports_array():
port_s = []
query_p = []
rcon_p = []
for k in list_config:
data_port = read_yaml(g="ark", m=k)
port_s.append(data_port['Port'])
query_p.append(data_port['QueryPort'])
rcon_p.append(data_port['RCONPort'])
return port_s, query_p, rcon_p
def ports(port, ports_arr, flag):
while True:
if not port:
if not ports_arr:
if flag == 0:
port = 7777
elif flag == 1:
port = 27015
elif flag == 2:
port = 23331
print_line(f"Port={port}", flag="CYAN")
return port
else:
port = max(ports_arr) + 2
print_line(f"Port={port}", flag="CYAN")
if port in ports_arr:
print_line("Порт уже занят", flag="RED")
port = check_int("Введите новый порт")
else:
return port
def config_cluster():
cluster_id = ""
cluster_dir_override = ""
count_cluster = check_int("""Укажите требуется ли кластер? default: Нет
1. Да
2. Нет
: """)
if count_cluster == 1:
cluster_server = True
while True:
cluster_id = input("Укажите id для кластера, любое сочетание символов: \n")
if cluster_id == '':
print_line("Введите символы: ", flag="CYAN")
else:
create_dir(dir_server_ark + cluster_id)
cluster_dir_override = (dir_server_ark + cluster_id)
break
else:
cluster_server = False
return cluster_server, cluster_id, cluster_dir_override
def config_nummap():
count_maps = check_int("Укажите количество карт: \n")
if count_maps == 0: # 0 возвращает check_int когда, ничего не было введено
count_maps = 1
return count_maps
def config_maps(i):
while True:
"""Проверка на выбор карты из списка"""
amount_map = check_int("""Выберите карту из списка указав номер
1. The Island
2. The Center
3. Scorched Earth
4. Ragnarok
5. Aberration
6. Extinction
7. Valguero
8. Genesis: Part 1
9. Crystal Isles
10. Genesis: Part 2
11. Lost Island
12. Fjordur
: """)
if amount_map == 0: # 0 возвращает check_int когда, ничего не было введено
amount_map = i + 1
if 0 < amount_map <= 12:
break
if list_config:
port_s, query_p, rcon_p = ports_array()
else:
port_s = query_p = rcon_p = []
if amount_map == 1:
map_s = "TheIsland"
elif amount_map == 2:
map_s = "TheCenter"
elif amount_map == 3:
map_s = "ScorchedEarth_P"
elif amount_map == 4:
map_s = "Ragnarok"
elif amount_map == 5:
map_s = "Aberration_P"
elif amount_map == 6:
map_s = "Extinction"
elif amount_map == 7:
map_s = "Valguero_P"
elif amount_map == 8:
map_s = "Genesis"
elif amount_map == 9:
map_s = "CrystalIsles"
elif amount_map == 10:
map_s = "Gen2"
elif amount_map == 11:
map_s = "LostIsland"
elif amount_map == 12:
map_s = "Fjordur"
else:
# Если вдруг каким-то боком проверка не отработает и не будет нужной цифры
map_s = 'TheIsland'
return map_s, port_s, query_p, rcon_p
def config_nameserver(map_s):
while True:
name_server = input("Укажите название Сервера: \n")
if name_server == "":
if map_s in list_config:
count = 1
new_name = map_s
while new_name in list_config:
new_name = f"{map_s}{str(count)}"
count += 1
name_server = new_name
break
else:
name_server = map_s
break
else:
if name_server in list_config:
choose_reconf = input("""Сервер существует. (по умолчанию - отмена)
1. Переименовать
2. Изменить настройки
3. Отмена
:""")
if choose_reconf == "" or choose_reconf == "3":
exit()
elif choose_reconf == "1":
new_nameserver = input("Укажите название Сервера: \n")
try:
if new_nameserver != name_server:
os.system(f"systemctl --user disable {dir_unit}ark_{name_server.lower()}.service")
os.system(f"rm {dir_unit}ark_{name_server.lower()}.service")
os.system(f"mv {dir_maps_ark}{name_server} {dir_maps_ark}{new_nameserver}")
return new_nameserver, True
else:
print_line("Вы ввели тоже имя.")
continue
except Exception:
print_line(Exception)
elif choose_reconf == "2":
config_ark(flag=False)
break
else:
print_line("Выберите ВОЗМОЖНОЕ действие")
else:
break
return name_server
def reconf_yaml():
name_server = input("Укажите название Сервера: \n")
return name_server
def config_ports(port_s):
port = check_int("Укажите порт сервера: ")
port_server = ports(port, port_s, 0)
return port_server
def config_query(query_p):
port = check_int("Укажите query порт сервера: ")
query_port = ports(port, query_p, 1)
return query_port
def config_rcon(rcon_p):
port = check_int("Укажите порт сервера: ")
rcon_port = ports(port, rcon_p, 2)
return rcon_port
def config_password():
password_server = input("Укажите пароль Сервера: \n")
return password_server
def config_adminpassword():
adminpassword_server = input("Укажите пароль администратора сервера: \n")
return adminpassword_server
def config_players():
max_players = check_int("Укажите максимальное количество игроков: \n")
if max_players == 0:
max_players = 70
return max_players
def config_listen():
print_line("Передавать сервер в глобальный список серверов steam?", flag="CYAN")
listen_server_amount = check_int("""\n
1. Да
2. Нет
:""")
if listen_server_amount == 1:
listen_server = True
elif listen_server_amount == 2:
listen_server = False
else:
listen_server = True
return listen_server
def config_ark(list_config=list_config, flag=False):
"""конфигурирование сервера арк"""
create_dir(dir_server_ark)
create_dir(dir_maps_ark)
id_mods_ark = ""
cluster_server, cluster_id, cluster_dir_override = config_cluster()
count_maps = config_nummap()
if list_config:
print_line("Существующие активные карты: ", flag="CYAN")
for i in list_config:
data = read_yaml(g="ark", m=i)
print_line(f"Карта - {data['map']} : Имя сервера {i}", flag="CYAN")
for i in range(count_maps):
map_s, port_s, query_p, rcon_p = config_maps(i)
if flag == True:
reconf_yaml()
break
else:
name_server, flag = config_nameserver(map_s)
if flag == False:
port_server = config_ports(port_s)
query_port = config_query(query_p)
rcon_port = config_rcon(rcon_p)
rcon_enabled = True
password_server = config_password()
adminpassword_server = config_adminpassword()
max_players = config_players()
listen_server = config_listen()
yaml_create("ark", "", cluster_server, map_s, name_server, port_server, query_port,
rcon_enabled, rcon_port, adminpassword_server, password_server, max_players,
id_mods_ark, cluster_id, cluster_dir_override, listen_server)
data = read_yaml(g="ark", m=map_s)
yaml_create("ark", data['ServerPath'], data['Cluster'], data['map'], name_server, data['Port'],
data['QueryPort'], data['RCONEnabled'], data['RCONPort'], data['ServerAdminPassword'],
data['ServerPassword'],data['MaxPlayers'], id_mods_ark, data['clusterid'], data['clusterdir'],
data['Listen'])
def config_7daystodie():
"""конфигурирование сервера 7 days to die"""
list_simvols = ("$", "@", "-", ".", "%", "{", "}", "+", "/")
create_dir(dir_server_7days)
config_7days = input("Введите имя конфига (serverconfig):\n")
if config_7days == "":
config_7days = "serverconfig"
elif config_7days == "serverconfig":
config_7days = "serverconfig"
elif config_7days in list_simvols:
print_line("Запрещённые символы", flag="RED")
else:
xml_parser()
systemd_unit_create("7Days", config_7days)
@hlna.command(help='Удаление конфигурации сервера')
@click.argument('g', nargs=1)
@click.option('-m', default='all', help="Название карты для запуска или all для запуска всех карт")
def config_backup(g, m):
if g.lower() == "ark":
name_server = choose_map(g, m)
if m == "all":
all_empty = True # флаг
for i in name_server:
data = read_yaml(g, m=i, flag=True)
if 'ark_backup' not in data or data['ark_backup'] == "" or data['ark_backup'] == "False":
all_empty = False # меняем флаг, если есть значение, которое не пустое
if all_empty:
config_backup_settings(g="ark") # если все значения пустые, то вызываем config_backup_settings
else:
pass # Тут должна быть проверка на то в каких серверах есть настройки резервного копирования, а в каких нет + вывод этой информации
# print_line("Уже настроен для карт: ", flag="CYAN")%
# name_servers = choose_map(g, m)
# print_line(name_servers)
# for i in name_servers:
# data = read_yaml(g="ark", i)
# print_line(f"Карта - {i} : Имя сервера {data['SessionName']}", flag="CYAN")
# choose_reconf = input("""Перенастроить? (по умолчанию) нет
# 1. Да
# 2. Нет
# :""")
# if choose_reconf == "" or "2":
# pass
# elif choose_reconf == "1":
# config_backup_settings(g)
# else:
# return config_backup(g)
# elif g == "7days":
# print_line("Пока не работает")
def config_backup_settings(g):
print_line(f"Входим в config_backup_settings - {g}")
if g == "ark":
data = read_yaml(g)
elif g == "7days":
print_line("Пока не работает")
def xml_parser():
"""добавить документацию"""
print_line("Я пока не умею парсить xml))", flag="RED")
def ini_parser():
"""добавить документацию"""
print_line("Я пока не умею парсить ini))", flag="RED")
def yaml_create(g, dir_server="", cluster_server="", map_s="", name_server="", port_server="", query_port="",
rcon_enabled="", rcon_port="", adminpassword_server="", password_server="", max_players="",
id_mods_ark="", cluster_id="", cluster_dir_override="", listen_server=""):
"""Создаёт на основании собранных данных yaml конфиг"""
if g == "ark":
path_yaml = dir_maps_ark + name_server
settings = [
{
'map': map_s,
'Cluster': cluster_server,
'SessionName': name_server,
'Port': port_server,
'QueryPort': query_port,
'RCONEnabled': rcon_enabled,
'RCONPort': rcon_port,
'ServerAdminPassword': adminpassword_server,
'ServerPassword': password_server,
'MaxPlayers': max_players,
'ModsId': id_mods_ark,
'Listen': listen_server,
'ServerPath': dir_server_ark,
'clusterid': cluster_id,
'clusterdir': cluster_dir_override
}
]
elif g == "path_server":
path_yaml = dir_config + "config"
settings = [
{
'path_server': dir_server
}
]
with open(path_yaml, 'w') as yamlfile:
yaml.dump(settings, yamlfile)
print_line(f"Конфиг {path_yaml} создан", flag="GREEN")
if g != "path_server":
systemd_unit_create(g, name_server)
def systemd_unit_create(g, name_server=list_config, config_7days=""):
"""Создаёт на основании yaml конфига systemd юнит"""
g = g.lower()
if g == "ark":
id_game = "376030"
data = read_yaml(g="ark", m=name_server)
ntff = "" if not data['Cluster'] else "-NoTransferFromFiltering"
unit_dir_server = dir_server_ark
dir_server_exec = f"{dir_server_ark}ShooterGame/Binaries/Linux/"
systemd_unit_exec = f"{dir_server_exec}ShooterGameServer {data['map']}?listen={data['Listen']}?SessionName={data['SessionName']}?Port={data['Port']}?QueryPort={data['QueryPort']}?RCONEnabled={data['RCONEnabled']}?RCONPort={data['RCONPort']}?ServerAdminPassword={data['ServerAdminPassword']}?MaxPlayers={data['MaxPlayers']}?GameModIds={data['ModsId']} -clusterid={data['clusterid']} -ClusterDirOverride={data['clusterdir']} {ntff}"
unit_file = f"{dir_unit}ark_{data['SessionName']}.service".lower()
elif g == "7Days":
id_game = "294420"
# сюда дописать обращение к xml_parser для получения уникального имени сервера
unit_dir_server = dir_server_7days
systemd_unit_exec = f"{dir_server_7days}startserver.sh -configfile={config_7days}.xml"
unit_file = f"{dir_unit}7days.service".lower()
else:
print_line("Пока доступно 2 игры: ark и 7Days")
unit_text = f'''[Unit]
Description={g}: Server
Wants=network-online.target
After=syslog.target network.target nss-lookup.target network-online.target
[Service]
ExecStartPre=/usr/bin/steamcmd +force_install_dir {unit_dir_server} +login anonymous +app_update {id_game} +quit
TimeoutStartSec=99999
ExecStart={systemd_unit_exec}
LimitNOFILE=100000
ExecReload=/bin/kill -s HUP $MAINPID
ExecStop=/bin/kill -s INT $MAINPID
[Install]
WantedBy=default.target
'''
with open(unit_file, 'w') as systemd_unit:
systemd_unit.write(unit_text)
unit_name = unit_file.split("/")[-1]
print_line(f"Юнит {unit_name} создан", flag="GREEN")
os.system('systemctl --user daemon-reload')
os.system(f"systemctl --user enable {unit_name}")
@hlna.command(help='Удаление конфигурации сервера')
@click.argument('g', nargs=1)
@click.option('-m', default='all', help="Название карты для запуска или all для запуска всех карт")
def remove(g, m):
if g == "ark":
name_server = choose_map(g, m)
os.remove(f"{dir_maps_ark}{name_server}")
@hlna.command(help='Скачивание и установка модов <hlna ark -m all -i 111111111>')
@click.argument('g', nargs=1)
@click.option('-m', default='all', help="Название карты для запуска или all для запуска всех карт")
@click.option("-i/-u", default=True, help="-i установить/обновить моды, -u удалить моды")
@click.argument('id_mods_ark', required=True, nargs=-1)
def mod(g, m, i, id_mods_ark):
g = g.lower()
if g == "ark":
check_exist_servers(g)
if not os.path.isdir(dir_mods_ark):
create_dir(dir_mods_ark)
id_mods_ark = id_mods_ark[0].split(',')
id_mods = ""
for id_mod in id_mods_ark:
id_mods += "," + id_mod
dir_mod_ark = f"{dir_mods_ark}/{id_mod}"
if not os.path.isfile(f"{dir_mod_ark}.mod"):
if i:
print_line(f"Скачиваем мод {id_mod}", flag="CYAN")
moddownload(g, m, id_mod, dir_mod_ark)
else:
os.system(f"rm -rf {dir_mod_ark}")
print_line(f"{dir_mod_ark} удалён", flag="CYAN")
os.system(f"rm {dir_mods_ark}/{id_mod}.mod")
print_line(f"{dir_mods_ark}/{id_mod}.mod удалён", flag="CYAN")
else:
print_line(f"Мод уже установлен", flag="CYAN")
modupdate(g, m, id_mod, dir_mod_ark)
name_server = choose_map(g, m)
id_mods = id_mods[1:]
id_mods_ark = id_mods
for i in name_server:
data = read_yaml(g="ark", m=i)
yaml_create("ark", data['ServerPath'], data['Cluster'], data['map'], data['SessionName'], data['Port'],
data['QueryPort'],
data['RCONEnabled'], data['RCONPort'], data['ServerAdminPassword'], data['ServerPassword'],
data['MaxPlayers'], id_mods_ark, data['clusterid'], data['clusterdir'], data['Listen'])
else:
print_line("Введите id модов через запятую без пробелов", flag="CYAN")
else:
print_line("Не поддерживаемая игра, возможно вы иммели ввиду ark?", flag="RED")
def modupdate(g, m, id_mod, dir_mod_ark):
g = g.lower()
if g == "ark":
print_line(f"Проверяем обновление мода {id_mod}", flag="CYAN")
if os.path.isfile(f"{dir_mod_ark}/appworkshop_346110.acf"):
with open(os.path.join(dir_mod_ark, f"appworkshop_346110.acf"), "r") as f:
content = f.readlines()
content = "".join(content)
locale_date = ""
for line in content.splitlines():
if '\t"WorkshopItemsInstalled"' in line:
for line in content.splitlines():
if f'\t\t"timeupdated"' in line:
locale_date = line.split('"')[3]
if '}' in line:
break
break
data = {
'itemcount': 1,
'publishedfileids[0]': id_mod
}
zapros = requests.post('http://api.steampowered.com/ISteamRemoteStorage/GetPublishedFileDetails/v1', data=data)
json_zapros = zapros.json()
steam_date = json_zapros['response']['publishedfiledetails'][0]['time_updated']
if int(steam_date) != int(locale_date):
print_line(f"Обновляем мод {id_mod}", flag="CYAN")
moddownload(g, m, id_mod, dir_mod_ark)
else:
print_line(f"Мод {id_mod} обновлен", flag="GREEN")
else:
print_line(f"Информация об обновлении не найдена. Переустанавливаем мод {id_mod}", flag="CYAN")
moddownload(g, m, id_mod, dir_mod_ark)
def modupdateall(g, m):
print_line("Проверяем обновление всех установленных модов", flag="CYAN")
if not os.path.isdir(dir_mods_ark):
create_dir(dir_mods_ark)
for file in os.listdir(dir_mods_ark): # Поменять на чтение списка модов из yaml
if os.path.isfile(os.path.join(dir_mods_ark, file)):
if file.endswith('.mod'):
id_mod = file.split(".")[0]
if id_mod == "111111111":
continue
dir_mod_ark = f"{dir_mods_ark}/{id_mod}"
modupdate(g, m, id_mod, dir_mod_ark)
def moddownload(g, m, id_mod, dir_mod_ark):
"""Распаковывает файлы мода и создаёт .mod файл для него"""
g = g.lower()
if g == "ark":
id_game_workshop = "346110"
dir_steam_workshop = f"{dir_workshop_ark}/content/{id_game_workshop}/{id_mod}/WindowsNoEditor"
dir_extract = dir_mod_ark
if id_mod == "111111111":
return
if os.path.isfile(f"{dir_workshop_ark}/appworkshop_{id_game_workshop}.acf"):
os.system(f"rm {dir_workshop_ark}/appworkshop_{id_game_workshop}.acf")
os.system(
f"steamcmd +force_install_dir {home_dir}/.local/share/Steam/ +login anonymous +workshop_download_item {id_game_workshop} {id_mod} +quit")
try:
for curdir, subdirs, files in os.walk(os.path.join(dir_steam_workshop)):
for file in files:
name, ext = os.path.splitext(file)
if ext == ".z":
src = os.path.join(curdir, file)
dst = os.path.join(curdir, name)
uncompressed = os.path.join(curdir, file + ".uncompressed_size")
unpack(src, dst)
print_line(f"[+] Извлечён {file}", flag="GREEN")
os.remove(src)
if os.path.isfile(uncompressed):
os.remove(uncompressed)
except Exception as e:
print_line(e, flag="RED")
print_line(f"[x] Unpacking .z files failed, aborting mod install", flag="RED")
return False
os.system(f"rm -rf {dir_mod_ark}")
os.system(f"mv -f {dir_steam_workshop} {dir_mod_ark}")
modname = subprocess.check_output(
['curl', '-s', f'https://steamcommunity.com/sharedfiles/filedetails/?id={id_mod}']).decode('utf-8')
modname = re.search(r'<div class="workshopItemTitle">(.+)</div>', modname)
modname = modname and modname.group(1)
if os.path.isfile(f"{dir_mod_ark}.mod"):
os.remove(f"{dir_mod_ark}.mod")
with open(f"{dir_extract}/mod.info", "rb") as modinfo:
data = modinfo.read()
mapnamelen = struct.unpack_from("<L", data, 0)[0]
mapname = data[4:mapnamelen + 3]
nummaps = struct.unpack_from("<L", data, mapnamelen + 4)[0]
pos = mapnamelen + 8
modname = (modname.encode() or mapname.decode()) + b'\x00'
modnamelen = len(modname)
modpath = b"../../../" + dir_shooter.encode() + b"/Content/Mods/" + id_mod.encode() + b'\x00'
modpathlen = len(modpath)
with open(f"{dir_mod_ark}.mod", "wb") as mod:
mod.write(struct.pack(f'<LLL{modnamelen}sL{modpathlen}sL', int(id_mod), 0, modnamelen, modname,
modpathlen, modpath, nummaps))
for mapnum in range(nummaps):
mapfilelen = struct.unpack_from("<L", data, pos)[0]
mapfile = data[mapnamelen + 12:mapnamelen + 12 + mapfilelen]
mod.write(struct.pack("<L%ds" % mapfilelen, mapfilelen, mapfile))
pos = pos + 4 + mapfilelen
mod.write(b"\x33\xFF\x22\xFF\x02\x00\x00\x00\x01")
if os.path.isfile(os.path.join(dir_extract, "modmeta.info")):
with open(os.path.join(dir_extract, "modmeta.info"), "rb") as f:
with open(f"{dir_extract}.mod", "ab") as f_out:
f_out.write(f.read())
else:
with open(f"{dir_mods_ark}.mod", "wb") as f_out:
f_out.write(b'\x01\x00\x00\x00\x08\x00\x00\x00ModType\x00\x02\x00\x00\x001\x00')
x = os.system(
f"mv {dir_workshop_ark}/appworkshop_{id_game_workshop}.acf {dir_mod_ark}/appworkshop_{id_game_workshop}.acf")
@hlna.command(help='Выключение/включение серверов (без удаления) <hlna switch -m all -d')
@click.argument('g', nargs=1)
@click.option("-m", required=True, help="Название cервера")
@click.option("-e/-d", default=True, help="-e активировать карты, -d деактивировать")
def switch(g, m, e): # добавить all
g = g.lower()
if g == "ark":
m = m.split(",")
if not os.path.isdir(dir_deactivated):
create_dir(dir_deactivated)
if e:
port_s, query_p, rcon_p = ports_array()
state_msg = "активных" if e else "не активных"
name_server = choose_map(g, m)
for i in name_server:
try:
data = read_yaml(g="ark", m=i, flag=not e)
if e: # добавить проверку занятости имени
if i in list_config:
data['SessionName'] = config_nameserver(i, flag=False)[-1]
data['Port'] = ports(data['Port'], port_s, e)
data['QueryPort'] = ports(data['QueryPort'], port_s, e)
data['RCONPort'] = ports(data['RCONPort'], port_s, e)
yaml_create("ark", data['ServerPath'], data['Cluster'], data['map'], data['SessionName'],
data['Port'],
data['QueryPort'], data['RCONEnabled'], data['RCONPort'], data['ServerAdminPassword'],
data['ServerPassword'], data['MaxPlayers'], data['ModsId'], data['clusterid'],
data['clusterdir'],
data['Listen'])
x = os.system(
f"rm {dir_deactivated}{i} >> {dir_logs}{date} 2>&1")
with open(f"{dir_logs}{date}.log", "a") as f:
f.write(f"[{t}] Сервер {i} перемещён из {state_msg}\n") # переписать эту залупу
else:
x = os.system(f"mv {dir_maps_ark}{i} {dir_deactivated} >> {dir_logs}{date} 2>&1")
with open(f"{dir_logs}{date}.log", "a") as f:
f.write(f"[{t}] Сервер {i} перемещён из {state_msg}\n")
if x == 0:
# start = "start" if e else "stop"
enable = "enable" if e else "disable"
os.system(f"systemctl --user {enable} ark_{i.lower()}")
print_line(f"Выполнено для сервера- {i}", flag="GREEN")
else:
print_line(f"Ошибка перемещения {i}", flag="RED")
except:
print_line("ошибка операции", flag="RED")
@hlna.command(help='Выводит статус настроеных серверов')
@click.argument('g', nargs=1)
@click.option("-m", default='all', help="Название cервера")
def status(g, m="all", list_config=list_config):
"""print_status делает вывод, flag - отвечает за показ активных/неактивных, под 7days надо будет дописать"""
def print_status(g, status_map, flag=True):
data = status_map
print_line(data["status"], flag=("YELLOW", "RED")[flag] if data['status'] == "Не запущен" else "GREEN")
print_line(f"""
Имя сервера: {data['SessionName']}
Карта: {data['map']}
Моды: {data['ModsId']}
Пароль: {data['ServerPassword']}
Кластер: {data['Cluster']}
Кластер id: {data['clusterid']}
Query порт: {data['QueryPort']}
Порт сервера: {data['Port']}
Rcon включен: {data['RCONEnabled']}
Rcon порт : {data['RCONPort']}
Максимальное кол-во игроков: {data['MaxPlayers']}
steam://connect/{get_external_ip()}:{data['QueryPort']}""", flag=("YELLOW","CYAN")[flag])
def get_param(g, list_con, flag=True):
for i in list_con:
data = read_yaml(g=g, m=i, flag=flag)
status_map["ark"][flag][data["SessionName"]] = data
status_map["ark"][flag][data["SessionName"]]['status'] = "Не запущен" if os.system(f"lsof -w -i :{data['Port']}") else "Запущен"
if list_config == [] and delist_config == []:
print_line("Сервера не сконфигурированы", flag="RED")
exit()
"Это скорее всего можно как то покрасивее записать, но пока так)"
status_map = {}
status_map["ark"] = {}
status_map["ark"][True] = {}
status_map["ark"][False] = {}
get_param("ark", list_config)
get_param("ark", delist_config, False)
if g == "ark":
name_servers = choose_map(g, m)
for i in name_servers:
print_status(g, status_map[g][True][i])
if delist_config != []:
x = input("Есть неактивные сервера, показать Y/n: ")
if x != "n":
for i in delist_config:
print_status(g, status_map[g][False][i], False)
return status_map
@hlna.command(help='Запуск, сконфигурированного сервера или кластера <hlna start ark -m all>')
@click.argument('g', nargs=1)
@click.option('-m', default='all', help="Название карты для запуска или all для запуска все карт")
def start(g, m):
"""Запускает сервер выбранной игры"""
# добавить проверку на ввод аргумента ark/7days если else: давать подсказку
# если нет конфигов, то выводим что серверов нет
g = g.lower()
if g == "ark":
modupdateall(g, m)
start_stop("start", g, m)
@hlna.command(help='Остановка, сконфигурированного сервера или кластера <hlna stop ark -m all>')
@click.argument('g', nargs=1)
@click.option('-m', default='all', help="Название карты для запуска или all для запуска все карт")
def stop(g, m):
g = g.lower()
if g == "ark":
modupdateall(g, m)
start_stop("stop", g, m)
@hlna.command(help='Перезапуск, сконфигурированного сервера или кластера <hlna restart ark -m all>')
@click.argument('g', nargs=1)
@click.option('-m', default='all', help="Название карты для запуска или all для запуска все карт")
def restart(g, m):
g = g.lower()
if g == "ark":
modupdateall(g, m)
start_stop("restart", g, m)
def check_exist_servers(g):
"""Проверяет наличие конфигов для активных карт"""
g = g.lower()
if g == "ark" and not list_config:
print_line("Нет сконфигурированных серверов", flag="RED") # добавить отсюда вилку на вопрос с конфигурацией
elif g == "7days":
print_line("7Days", flag="CYAN")
else:
return 1
def start_stop(action, g, m):
"""Функция изменения статусов сервера"""
g = g.lower()
if g == "ark":
x = check_exist_servers(g)
if x:
name_servers = choose_map(g, m) if m != 'all' else list_config
for i in name_servers:
data = read_yaml(g="ark", m=i, flag=True)
if action == "stop" or action == "restart":
rcon_local(i, "SaveWorld")
x = os.system(f"systemctl --user {action} ark_{data['SessionName'].lower()}.service")
if x == 0:
print_line(f"Готово {action} для {g} {i}", flag="GREEN")
elif g == "7days":
x = os.system(f"systemctl --user {action} 7days.service")
if x == 0:
print_line(f"Готово {action} для {m}", flag="GREEN")
else:
print_line("доступные игры: ark и 7days")
def read_yaml(g="", m="", flag=True):
"""Читает конфиги активных или неактивных карт в зависимости от флага и отдаёт данные туда где их запросили"""
g = g.lower()
if g == "ark":
if m == "all":
print_line("Не правильный вызов yaml, должен вызываться из цикла")
path_yaml = f"{dir_maps_ark}{m}" if flag else f"{dir_deactivated}{m}"
elif g == "path_server":
path_yaml = dir_config + "config"
with open(path_yaml, "r") as yamlfile:
data = yaml.load(yamlfile, Loader=yaml.FullLoader)
return data[0] # возвращаем словарь со всеми значениями
def choose_map(g, m, list_config=list_config):
"""Функция выбора карт"""
g = g.lower()
new_arr = []
if g == "ark":
dict_mapname = {}
dict_allmapname = []
print_line(list_config)
for i in list_config:
data = read_yaml(g="ark", m=i)
dict_mapname[data['SessionName']] = data['map']
dict_allmapname.append(data['SessionName'])
name_servers = []
for ns, v in dict_mapname.items():
if v in m:
name_servers.append(ns)
if list_config != []: # Перенести выше для проверки наличия конфигов
if m == "all":
new_arr = dict_allmapname
print_line(f"Выполняется для карт(-ы) {new_arr}", flag="CYAN")
else:
if name_servers:
name_servers = sorted(name_servers)
print_line('Найдены сервера с этой картой', flag="CYAN")
for i, map in enumerate(name_servers):
print_line(f"{i + 1}) {map}", flag="CYAN")
print_line(f"{i + 2} Все", flag="CYAN")
x = None
if i != 0:
while True:
try:
x = input("Выберите сервер из списка, либо несколько через запятую: ").split(',')
x = [int(j) for j in x]
if i + 2 in x:
return name_servers
break
except:
print_line("Неправильный ввод", flag="RED")
else:
x = [1]
for i in x:
new_arr.append(name_servers[i - 1])
print_line(f"Выбранные сервера: {name_servers}", flag="CYAN")
else:
print_line("Не найдено серверов с картой")
return new_arr
@hlna.command(help='Отправка команд на игровой сервер через rcon <rcon SaveWorld -m all>')
@click.argument('c', nargs=1)
@click.option('-m', required=True, help="Название карты для применения rcon команды")
def rcon(m, c):
rcon_local(m, c)
def rcon_local(m, c):
try:
dict_mapname = {}
dict_adminpwd = {}
if list_config:
rcon_ports = []
for i in list_config:
data = read_yaml(g="ark", m=i)
dict_mapname[data['RCONPort']] = data['map']
dict_adminpwd[data['RCONPort']] = data['ServerAdminPassword']
if m == "all":
for rcon_p in dict_mapname:
rcon_ports.append(rcon_p)
else:
for rcon_p, name_map in dict_mapname.items():
if name_map in m:
rcon_ports.append(rcon_p)
for port in rcon_ports:
passwd = dict_adminpwd[port]
with Client('127.0.0.1', port, passwd=str(passwd)) as client:
response = client.run(c)
print_line(f"Rcon выполнен {response} {dict_mapname[port]}", flag="GREEN")
else:
pass
except:
print_line(f"Ошибка отправки команды {c} в {m}", flag="RED")
def zero(x=""):
"""Потом пригодится (нет)"""
return ""
if not os.path.exists(dir_config + "config"):
dir_server = path_server()
else:
data = read_yaml(g="path_server")
if data['path_server'] == "":
path_server()
else:
dir_server = data['path_server']
dir_unit = f"{home_dir}/.config/systemd/user/"
dir_logs = f"{dir_config}logs/"
dir_server_ark = f"{dir_server}ARK/"
dir_ark_save = f"{dir_server_ark}ShooterGame/Saved/SavedArks/"
dir_workshop_ark = f"{home_dir}/.local/share/Steam/steamapps/workshop"
dir_shooter = "ShooterGame"
dir_mods_ark = f"{dir_server_ark}ShooterGame/Content/Mods"
dir_server_7days = f"{dir_server}/7Days/"
now = datetime.datetime.now()
date = now.strftime("%Y-%m-%d")
t = now.strftime("%H:%M:%S")
create_dir(dir_server)
create_dir(dir_unit)
create_dir(dir_logs)
class HlnaApp(QtWidgets.QMainWindow, hlnaui.Ui_mainWindow):
def __init__(self):
super().__init__()
self.setupUi(self)
def hlnag():
if len(sys.argv) > 1:
hlna()
else:
app = QtWidgets.QApplication(sys.argv)
window = HlnaApp()
window.show()
sys.exit(app.exec())
if __name__ == '__main__':
hlnag()